dregs 1.0.0

//! The Dregs client, and the transport, retries, and error mapping behind every call.

use std::fmt;
use std::sync::Arc;
use std::time::Duration;

use reqwest::header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, HeaderMap, USER_AGENT};
use reqwest::{Method, StatusCode};
use serde_json::Value;

use percent_encoding::{AsciiSet, NON_ALPHANUMERIC, utf8_percent_encode};

use crate::devices::Devices;
use crate::error::{ApiError, ApiErrorKind, Error, Result};
use crate::escalations::Escalations;
use crate::event::Event;
use crate::events::Events;
use crate::groups::Groups;
use crate::identities::Identities;
use crate::models::TrackResult;

/// The API root used when none is configured.
pub const DEFAULT_BASE_URL: &str = "https://dregs.com/api";

/// How long a request may take before it is abandoned, when none is configured.
pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);

/// How many times a failed request is retried, when no limit is configured.
pub const DEFAULT_MAX_RETRIES: u32 = 2;

/// The environment variable the secret key is read from.
pub const SECRET_KEY_ENV: &str = "DREGS_SECRET_KEY";

/// The environment variable the base URL is read from.
pub const BASE_URL_ENV: &str = "DREGS_BASE_URL";

/// Statuses worth another attempt. 429 and 5xx are transient by definition; 408 shows up in front
/// of some proxies.
const RETRY_STATUSES: [u16; 6] = [408, 429, 500, 502, 503, 504];

/// The longest a `Retry-After` header is allowed to hold a retry back.
const MAX_RETRY_AFTER: Duration = Duration::from_secs(60);

/// Body-level statuses Dregs uses on `POST /api/events`.
const STATUS_RATE_LIMITED: &str = "rate_limited";
const STATUS_QUOTA_EXCEEDED: &str = "quota_exceeded";

const SDK_USER_AGENT: &str = concat!("dregs-rust/", env!("CARGO_PKG_VERSION"));

/// Everything but the unreserved characters of RFC 3986, so an id is always a single path segment
/// and a query value can never break out of its parameter.
const UNRESERVED: &AsciiSet = &NON_ALPHANUMERIC
    .remove(b'-')
    .remove(b'.')
    .remove(b'_')
    .remove(b'~');

/// A Dregs client.
///
/// ```no_run
/// use dregs::{Client, Event};
///
/// # async fn run() -> dregs::Result<()> {
/// let client = Client::from_env()?;
///
/// client
///     .track(Event::new("user.signup", "user_12345").data(serde_json::json!({ "plan": "pro" })))
///     .await?;
///
/// let scores = client.identities().scores("user_12345").await?;
/// # Ok(())
/// # }
/// ```
///
/// The secret key is the one under **Settings → Credentials** in the dashboard that starts
/// `sk_`, not the `pk_` public key the browser tracker uses.
///
/// The client holds a connection pool, so build one and keep it rather than making a new one per
/// request. Cloning is cheap and every clone shares the pool, so hand clones to tasks and
/// handlers freely.
#[derive(Clone)]
pub struct Client {
    inner: Arc<Inner>,
}

struct Inner {
    secret_key: String,
    base_url: String,
    timeout: Duration,
    max_retries: u32,
    http: reqwest::Client,
}

impl Client {
    /// A client authenticating with `secret_key`, with every other setting at its default.
    ///
    /// The base URL still comes from `$DREGS_BASE_URL` when it is set.
    pub fn new(secret_key: impl Into<String>) -> Result<Self> {
        Client::builder().secret_key(secret_key).build()
    }

    /// A client whose secret key comes from `$DREGS_SECRET_KEY`.
    pub fn from_env() -> Result<Self> {
        Client::builder().build()
    }

    /// A builder for a client with settings other than the defaults.
    pub fn builder() -> ClientBuilder {
        ClientBuilder::default()
    }

    /// The API root every request is built against.
    pub fn base_url(&self) -> &str {
        &self.inner.base_url
    }

    /// How many times a failed request is retried before the error is returned.
    pub fn max_retries(&self) -> u32 {
        self.inner.max_retries
    }

    /// Read identities, their scores, and their analysis.
    pub fn identities(&self) -> Identities<'_> {
        Identities::new(self)
    }

    /// Read and manage groups and their members.
    pub fn groups(&self) -> Groups<'_> {
        Groups::new(self)
    }

    /// Read the events Dregs has recorded.
    pub fn events(&self) -> Events<'_> {
        Events::new(self)
    }

    /// Read devices, and disregard them.
    pub fn devices(&self) -> Devices<'_> {
        Devices::new(self)
    }

    /// Read and work escalations.
    pub fn escalations(&self) -> Escalations<'_> {
        Escalations::new(self)
    }

    /// Records a backend event against an identity.
    ///
    /// ```no_run
    /// # async fn run(client: dregs::Client) -> dregs::Result<()> {
    /// use dregs::Event;
    /// use serde_json::json;
    ///
    /// let result = client
    ///     .track(
    ///         Event::new("user.signup", "user_12345")
    ///             .data(json!({ "plan": "pro", "referrer": "partner-x" }))
    ///             .identity_data(json!({ "email": "ada@example.com", "name": "Ada Lovelace" })),
    ///     )
    ///     .await?;
    ///
    /// if !result.accepted() {
    ///     eprintln!("Dregs did not record the event (status {:?})", result.status);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// Check [`TrackResult::accepted`] to confirm Dregs recorded the event. Failures that are
    /// yours to act on come back as errors: [`QuotaExceeded`](ApiErrorKind::QuotaExceeded) when
    /// the account is over its monthly event limit, [`RateLimited`](ApiErrorKind::RateLimited)
    /// when the credential is ingesting too fast, [`Authentication`](ApiErrorKind::Authentication)
    /// when the secret key was not recognized, and [`BadRequest`](ApiErrorKind::BadRequest) when
    /// the event was malformed.
    pub async fn track(&self, event: Event) -> Result<TrackResult> {
        let body = event.into_body()?;
        let payload = self
            .request(Method::POST, "/events", &[], Some(&body))
            .await?;

        Ok(TrackResult::from_value(object_or_null(payload)))
    }

    /// Sends a request, retrying what is worth retrying, and returns the parsed JSON body.
    ///
    /// `query` is sent as given, so leave out the parameters that are not set.
    pub(crate) async fn request(
        &self,
        method: Method,
        path: &str,
        query: &[(&str, String)],
        body: Option<&Value>,
    ) -> Result<Option<Value>> {
        let mut url = format!("{}/{}", self.inner.base_url, path.trim_start_matches('/'));
        let body =
            body.map(|body| serde_json::to_vec(body).expect("a JSON value always serializes"));

        for (index, (name, value)) in query.iter().enumerate() {
            url.push(if index == 0 { '?' } else { '&' });
            url.push_str(name);
            url.push('=');
            url.extend(utf8_percent_encode(value, UNRESERVED));
        }
        let mut attempt = 0;

        loop {
            let error = match self.send(method.clone(), &url, body.clone()).await {
                Ok(response) => match process(response) {
                    Ok(payload) => return Ok(payload),
                    Err(error) => error,
                },
                Err(source) if source.is_timeout() => Error::Timeout {
                    url: url.clone(),
                    source,
                },
                Err(source) => Error::Connection {
                    url: url.clone(),
                    source,
                },
            };

            let (status, retry_after) = match &error {
                Error::Api(api) => (Some(api.status), api.retry_after),
                _ => (None, None),
            };

            if !self.should_retry(attempt, status) {
                return Err(error);
            }

            tokio::time::sleep(backoff(attempt, retry_after)).await;
            attempt += 1;
        }
    }

    /// One attempt, read to the end, so a connection dropped mid-body is retried like any other
    /// connection failure.
    async fn send(
        &self,
        method: Method,
        url: &str,
        body: Option<Vec<u8>>,
    ) -> std::result::Result<RawResponse, reqwest::Error> {
        let mut request = self
            .inner
            .http
            .request(method, url)
            .timeout(self.inner.timeout)
            .header(AUTHORIZATION, format!("Bearer {}", self.inner.secret_key))
            .header(ACCEPT, "application/json")
            .header(USER_AGENT, SDK_USER_AGENT);

        if let Some(body) = body {
            request = request.header(CONTENT_TYPE, "application/json").body(body);
        }

        let response = request.send().await?;
        let status = response.status();
        let headers = response.headers().clone();
        let body = response.bytes().await?;

        Ok(RawResponse {
            status,
            headers,
            body: body.to_vec(),
        })
    }

    fn should_retry(&self, attempt: u32, status: Option<u16>) -> bool {
        if attempt >= self.inner.max_retries {
            return false;
        }

        status.is_none_or(|status| RETRY_STATUSES.contains(&status))
    }
}

impl fmt::Debug for Client {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Client")
            .field("base_url", &self.inner.base_url)
            .field("timeout", &self.inner.timeout)
            .field("max_retries", &self.inner.max_retries)
            .finish_non_exhaustive()
    }
}

/// Builds a [`Client`] with settings other than the defaults.
///
/// ```no_run
/// use std::time::Duration;
///
/// # fn run() -> dregs::Result<()> {
/// let client = dregs::Client::builder()
///     .secret_key(std::env::var("DREGS_SECRET_KEY").unwrap_or_default())
///     .timeout(Duration::from_secs(5))
///     .max_retries(5)
///     .build()?;
/// # Ok(())
/// # }
/// ```
#[derive(Default)]
pub struct ClientBuilder {
    secret_key: Option<String>,
    base_url: Option<String>,
    timeout: Option<Duration>,
    max_retries: Option<u32>,
    http_client: Option<reqwest::Client>,
}

impl ClientBuilder {
    /// Your credential's secret key. Defaults to `$DREGS_SECRET_KEY`.
    pub fn secret_key(mut self, secret_key: impl Into<String>) -> Self {
        self.secret_key = Some(secret_key.into());
        self
    }

    /// The API root. Defaults to `$DREGS_BASE_URL`, then `https://dregs.com/api`.
    pub fn base_url(mut self, base_url: impl Into<String>) -> Self {
        self.base_url = Some(base_url.into());
        self
    }

    /// How long a request may take before it is abandoned. Defaults to ten seconds.
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.timeout = Some(timeout);
        self
    }

    /// How many times to retry a failed request. Defaults to two.
    ///
    /// Retries cover connection failures, timeouts, 429s, and 5xx, with exponential backoff and
    /// jitter; the `Retry-After` header wins when the server sends one. Pass 0 to handle failures
    /// yourself.
    pub fn max_retries(mut self, max_retries: u32) -> Self {
        self.max_retries = Some(max_retries);
        self
    }

    /// A `reqwest::Client` to use instead of the one this client would build, for callers who
    /// need their own proxies, TLS settings, or connection pool. The configured
    /// [`timeout`](Self::timeout) still applies to each request.
    pub fn http_client(mut self, http_client: reqwest::Client) -> Self {
        self.http_client = Some(http_client);
        self
    }

    /// Builds the client.
    ///
    /// Returns [`Error::InvalidArgument`] when there is no secret key, or when the key given is a
    /// public key.
    pub fn build(self) -> Result<Client> {
        let secret_key = self
            .secret_key
            .filter(|key| !key.is_empty())
            .or_else(|| env(SECRET_KEY_ENV))
            .ok_or_else(|| {
                Error::InvalidArgument(format!(
                    "No Dregs secret key. Pass one to Client::new or set the {SECRET_KEY_ENV} \
                     environment variable. You will find your credential's secret key under \
                     Settings -> Credentials in the Dregs dashboard."
                ))
            })?;

        if secret_key.starts_with("pk_") {
            return Err(Error::InvalidArgument(
                "That is a public key. The public key is for the browser tracker and cannot read \
                 identities or scores; this SDK needs the secret key from the same credential, \
                 which starts with 'sk_'."
                    .into(),
            ));
        }

        let base_url = self
            .base_url
            .filter(|url| !url.is_empty())
            .or_else(|| env(BASE_URL_ENV))
            .unwrap_or_else(|| DEFAULT_BASE_URL.into())
            .trim_end_matches('/')
            .to_owned();

        let http = match self.http_client {
            Some(http) => http,
            None => reqwest::Client::builder().build().map_err(|error| {
                Error::InvalidArgument(format!("Could not build the HTTP client: {error}"))
            })?,
        };

        Ok(Client {
            inner: Arc::new(Inner {
                secret_key,
                base_url,
                timeout: self.timeout.unwrap_or(DEFAULT_TIMEOUT),
                max_retries: self.max_retries.unwrap_or(DEFAULT_MAX_RETRIES),
                http,
            }),
        })
    }
}

impl fmt::Debug for ClientBuilder {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ClientBuilder")
            .field("base_url", &self.base_url)
            .field("timeout", &self.timeout)
            .field("max_retries", &self.max_retries)
            .finish_non_exhaustive()
    }
}

struct RawResponse {
    status: StatusCode,
    headers: HeaderMap,
    body: Vec<u8>,
}

/// Turns a response into parsed JSON, or the matching error.
fn process(response: RawResponse) -> Result<Option<Value>> {
    let payload: Option<Value> = if response.body.is_empty() {
        None
    } else {
        serde_json::from_slice(&response.body).ok()
    };

    let request_id = header(&response.headers, "x-request-id");

    if response.status.as_u16() >= 400 {
        let status = response.status.as_u16();
        let kind = ApiErrorKind::from_status(status);
        let message = payload
            .as_ref()
            .and_then(message_from)
            .or_else(|| response.status.canonical_reason().map(str::to_owned))
            .unwrap_or_else(|| "Request failed".into());

        return Err(Error::Api(ApiError {
            kind,
            status,
            message,
            body: payload,
            request_id,
            retry_after: if kind == ApiErrorKind::RateLimited {
                retry_after(&response.headers)
            } else {
                None
            },
        }));
    }

    // An older API build reported both of these as HTTP 200 with the outcome in the body. Reading
    // the body as well as the status keeps this SDK correct against either.
    let body_status = payload
        .as_ref()
        .and_then(|payload| payload.get("status"))
        .and_then(Value::as_str);

    match body_status {
        Some(STATUS_RATE_LIMITED) => Err(Error::Api(ApiError {
            kind: ApiErrorKind::RateLimited,
            status: 429,
            message: "Ingestion rate limit exceeded for this credential.".into(),
            retry_after: retry_after(&response.headers),
            body: payload,
            request_id,
        })),
        Some(STATUS_QUOTA_EXCEEDED) => Err(Error::Api(ApiError {
            kind: ApiErrorKind::QuotaExceeded,
            status: 402,
            message: "The account is over its monthly event limit.".into(),
            retry_after: None,
            body: payload,
            request_id,
        })),
        _ => Ok(payload),
    }
}

/// How long to wait before attempt `attempt + 1`.
///
/// `Retry-After` wins when the server sent one. Otherwise this is exponential with full jitter,
/// which keeps a fleet of workers that all hit the limit at once from retrying in lockstep.
fn backoff(attempt: u32, retry_after: Option<Duration>) -> Duration {
    if let Some(retry_after) = retry_after {
        return retry_after.min(MAX_RETRY_AFTER);
    }

    let ceiling = (0.5 * 2f64.powi(attempt.min(16) as i32)).min(8.0);

    Duration::from_secs_f64(fastrand::f64() * ceiling)
}

fn header(headers: &HeaderMap, name: &str) -> Option<String> {
    headers
        .get(name)
        .and_then(|value| value.to_str().ok())
        .map(str::to_owned)
}

/// Reads `Retry-After` as seconds.
///
/// The header also allows an HTTP date, which is rare enough here that falling back to the
/// client's own backoff is better than dragging in a date parser.
fn retry_after(headers: &HeaderMap) -> Option<Duration> {
    header(headers, "retry-after")?
        .trim()
        .parse::<f64>()
        .ok()
        .filter(|seconds| seconds.is_finite() && *seconds >= 0.0)
        .map(Duration::from_secs_f64)
}

fn message_from(payload: &Value) -> Option<String> {
    ["message", "error", "status"].into_iter().find_map(|key| {
        payload
            .get(key)
            .and_then(Value::as_str)
            .filter(|value| !value.is_empty())
            .map(str::to_owned)
    })
}

/// Escapes `value` as a single path segment, refusing an empty one before any request.
///
/// Ids are the caller's own and routinely contain characters that need escaping (an email address
/// being the common one).
pub(crate) fn segment(value: &str, what: &str) -> Result<String> {
    if value.is_empty() {
        return Err(Error::InvalidArgument(format!("{what} is required.")));
    }

    Ok(utf8_percent_encode(value, UNRESERVED).to_string())
}

/// The payload when it is a JSON object, and `Null` otherwise, for the models to read.
pub(crate) fn object_or_null(payload: Option<Value>) -> Value {
    payload.filter(Value::is_object).unwrap_or(Value::Null)
}

fn env(name: &str) -> Option<String> {
    std::env::var(name).ok().filter(|value| !value.is_empty())
}

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

    #[test]
    fn retry_after_wins_but_is_capped() {
        assert_eq!(
            backoff(0, Some(Duration::from_secs(2))),
            Duration::from_secs(2)
        );
        assert_eq!(backoff(0, Some(Duration::from_secs(600))), MAX_RETRY_AFTER);
    }

    #[test]
    fn jittered_backoff_stays_under_its_ceiling() {
        for attempt in 0..40 {
            let ceiling = (0.5 * 2f64.powi(attempt.min(16) as i32)).min(8.0);

            assert!(backoff(attempt, None).as_secs_f64() <= ceiling);
        }
    }

    #[test]
    fn the_secret_key_stays_out_of_debug_output() {
        let client = Client::builder()
            .secret_key("sk_abcdefghQijklmQabcdefghijklmn")
            .base_url("https://api.test.invalid/api")
            .build()
            .unwrap();

        assert!(!format!("{client:?}").contains("sk_"));
    }
}