linear_api/

client.rs

//! [`LinearClient`]: connection pool, auth, the retry/rate-limit-aware
//! execute loop, and the rate-limit budget snapshot.

use std::sync::{Arc, PoisonError, RwLock};
use std::time::Duration;

use reqwest::StatusCode;
use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderValue, RETRY_AFTER};
use secrecy::{ExposeSecret, SecretString};
use serde::Serialize;
use serde::de::DeserializeOwned;

use crate::error::{Error, GraphQlError, Result};

const DEFAULT_ENDPOINT: &str = "https://api.linear.app/graphql";
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
const DEFAULT_USER_AGENT: &str = concat!("linear-api-rs/", env!("CARGO_PKG_VERSION"));
/// Exponential backoff is capped here; rate-limit waits use the
/// header-derived wait instead.
const MAX_BACKOFF: Duration = Duration::from_secs(8);

/// Async client for the Linear GraphQL API.
///
/// Cheap to clone (an [`Arc`] around the connection pool); build one per
/// process and clone it across tasks.
#[derive(Debug, Clone)]
pub struct LinearClient {
    inner: Arc<ClientInner>,
}

#[derive(Debug)]
struct ClientInner {
    http: reqwest::Client,
    endpoint: String,
    api_key: SecretString,
    retry: RetryConfig,
    last_rate_limit: RwLock<Option<RateLimitInfo>>,
}

/// Retry policy. Defaults are safe for non-idempotent mutations: Linear has
/// no idempotency keys, so mutations are **not** retried on post-send
/// transport errors or 5xx unless
/// [`retry_mutations_on_transient`](RetryConfig::retry_mutations_on_transient)
/// is opted into. Rate-limit rejections happen before execution and are
/// always retried (within [`max_rate_limit_wait`](RetryConfig::max_rate_limit_wait)).
#[derive(Debug, Clone)]
pub struct RetryConfig {
    /// Total attempts including the first (default 3 = 1 initial + 2 retries).
    pub max_attempts: u32,
    /// Base for exponential backoff with full jitter (default 250ms,
    /// doubling per attempt, capped at 8s).
    pub base_backoff: Duration,
    /// Longest rate-limit wait to sit out in-process (default 30s). Longer
    /// waits fail fast with [`Error::RateLimited`] instead of parking the
    /// caller.
    pub max_rate_limit_wait: Duration,
    /// Also retry mutations on post-send transport errors / 5xx (default
    /// `false`; a timed-out `issueCreate` may have landed — blind retries
    /// can double-create).
    pub retry_mutations_on_transient: bool,
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_attempts: 3,
            base_backoff: Duration::from_millis(250),
            max_rate_limit_wait: Duration::from_secs(30),
            retry_mutations_on_transient: false,
        }
    }
}

/// Snapshot of Linear's rate-limit budget headers, parsed from every
/// response. The reset headers are UTC **epoch milliseconds** on the wire.
///
/// Never hardcode budgets — Linear's documented numbers are inconsistent;
/// these headers are the source of truth.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct RateLimitInfo {
    /// Requests allowed per window (`X-RateLimit-Requests-Limit`).
    pub requests_limit: Option<u64>,
    /// Requests remaining in the window (`X-RateLimit-Requests-Remaining`).
    pub requests_remaining: Option<u64>,
    /// When the request budget resets (`X-RateLimit-Requests-Reset`).
    pub requests_reset: Option<time::OffsetDateTime>,
    /// Complexity consumed by the last query (`X-Complexity`).
    pub complexity_last_query: Option<u64>,
    /// Complexity allowed per window (`X-RateLimit-Complexity-Limit`).
    pub complexity_limit: Option<u64>,
    /// Complexity remaining in the window (`X-RateLimit-Complexity-Remaining`).
    pub complexity_remaining: Option<u64>,
    /// When the complexity budget resets (`X-RateLimit-Complexity-Reset`).
    pub complexity_reset: Option<time::OffsetDateTime>,
    /// Endpoint-specific limit name (`X-RateLimit-Endpoint-Name`), when present.
    pub endpoint_name: Option<String>,
    /// Endpoint-specific requests remaining
    /// (`X-RateLimit-Endpoint-Requests-Remaining`), when present.
    pub endpoint_requests_remaining: Option<u64>,
}

impl RateLimitInfo {
    fn from_headers(headers: &HeaderMap) -> Self {
        Self {
            requests_limit: header_u64(headers, "x-ratelimit-requests-limit"),
            requests_remaining: header_u64(headers, "x-ratelimit-requests-remaining"),
            requests_reset: header_epoch_ms(headers, "x-ratelimit-requests-reset"),
            complexity_last_query: header_u64(headers, "x-complexity"),
            complexity_limit: header_u64(headers, "x-ratelimit-complexity-limit"),
            complexity_remaining: header_u64(headers, "x-ratelimit-complexity-remaining"),
            complexity_reset: header_epoch_ms(headers, "x-ratelimit-complexity-reset"),
            endpoint_name: headers
                .get("x-ratelimit-endpoint-name")
                .and_then(|v| v.to_str().ok())
                .map(str::to_owned),
            endpoint_requests_remaining: header_u64(
                headers,
                "x-ratelimit-endpoint-requests-remaining",
            ),
        }
    }

    fn is_empty(&self) -> bool {
        self.requests_limit.is_none()
            && self.requests_remaining.is_none()
            && self.requests_reset.is_none()
            && self.complexity_last_query.is_none()
            && self.complexity_limit.is_none()
            && self.complexity_remaining.is_none()
            && self.complexity_reset.is_none()
            && self.endpoint_name.is_none()
            && self.endpoint_requests_remaining.is_none()
    }
}

fn header_u64(headers: &HeaderMap, name: &str) -> Option<u64> {
    headers.get(name)?.to_str().ok()?.trim().parse().ok()
}

fn header_epoch_ms(headers: &HeaderMap, name: &str) -> Option<time::OffsetDateTime> {
    let ms: i128 = headers.get(name)?.to_str().ok()?.trim().parse().ok()?;
    time::OffsetDateTime::from_unix_timestamp_nanos(ms.checked_mul(1_000_000)?).ok()
}

/// Builder for [`LinearClient`]. Obtain via [`LinearClient::builder`].
#[derive(Debug, Default)]
pub struct LinearClientBuilder {
    api_key: Option<SecretString>,
    endpoint: Option<String>,
    timeout: Option<Duration>,
    connect_timeout: Option<Duration>,
    user_agent: Option<String>,
    retry: Option<RetryConfig>,
}

impl LinearClientBuilder {
    /// Sets the Linear API key (required). Keys look like `lin_api_…` and
    /// are sent as a raw `Authorization` header — no `Bearer` prefix.
    pub fn api_key(mut self, key: impl Into<SecretString>) -> Self {
        self.api_key = Some(key.into());
        self
    }

    /// Overrides the GraphQL endpoint (default
    /// `https://api.linear.app/graphql`). Useful for mock servers.
    pub fn endpoint(mut self, url: impl Into<String>) -> Self {
        self.endpoint = Some(url.into());
        self
    }

    /// Total per-request timeout (default 30s).
    pub fn timeout(mut self, d: Duration) -> Self {
        self.timeout = Some(d);
        self
    }

    /// Connection timeout (default 10s).
    pub fn connect_timeout(mut self, d: Duration) -> Self {
        self.connect_timeout = Some(d);
        self
    }

    /// `User-Agent` header (default `linear-api-rs/{version}`).
    pub fn user_agent(mut self, ua: impl Into<String>) -> Self {
        self.user_agent = Some(ua.into());
        self
    }

    /// Retry policy (default [`RetryConfig::default`]).
    pub fn retry(mut self, cfg: RetryConfig) -> Self {
        self.retry = Some(cfg);
        self
    }

    /// Builds the client. Fails with [`Error::Config`] when the API key is
    /// missing, the endpoint URL is invalid, or the key contains characters
    /// not permitted in an HTTP header.
    pub fn build(self) -> Result<LinearClient> {
        let api_key = self.api_key.ok_or_else(|| {
            Error::Config("no API key provided; set one with LinearClientBuilder::api_key".into())
        })?;
        HeaderValue::from_str(api_key.expose_secret()).map_err(|_| {
            Error::Config("API key contains characters not permitted in an HTTP header".into())
        })?;
        let endpoint = self.endpoint.unwrap_or_else(|| DEFAULT_ENDPOINT.to_owned());
        url::Url::parse(&endpoint)
            .map_err(|e| Error::Config(format!("invalid endpoint URL {endpoint:?}: {e}")))?;
        let http = reqwest::Client::builder()
            .connect_timeout(self.connect_timeout.unwrap_or(DEFAULT_CONNECT_TIMEOUT))
            .timeout(self.timeout.unwrap_or(DEFAULT_TIMEOUT))
            .user_agent(
                self.user_agent
                    .unwrap_or_else(|| DEFAULT_USER_AGENT.to_owned()),
            )
            .build()
            .map_err(|e| Error::Config(format!("failed to build HTTP client: {e}")))?;
        Ok(LinearClient {
            inner: Arc::new(ClientInner {
                http,
                endpoint,
                api_key,
                retry: self.retry.unwrap_or_default(),
                last_rate_limit: RwLock::new(None),
            }),
        })
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OpKind {
    Query,
    Mutation,
}

impl LinearClient {
    /// Starts building a client.
    pub fn builder() -> LinearClientBuilder {
        LinearClientBuilder::default()
    }

    /// Builds a client with all defaults and the given API key.
    pub fn new(api_key: impl Into<SecretString>) -> Result<Self> {
        Self::builder().api_key(api_key).build()
    }

    /// Builds a client from the `LINEAR_API_KEY` environment variable.
    /// Fails with [`Error::Config`] when unset.
    pub fn from_env() -> Result<Self> {
        let key = std::env::var("LINEAR_API_KEY")
            .map_err(|_| Error::Config("LINEAR_API_KEY environment variable is not set".into()))?;
        Self::new(key)
    }

    /// The most recent rate-limit budget snapshot observed on any response.
    pub fn last_rate_limit(&self) -> Option<RateLimitInfo> {
        self.inner
            .last_rate_limit
            .read()
            .unwrap_or_else(PoisonError::into_inner)
            .clone()
    }

    /// Executes one GraphQL **query** with retry (transient failures and
    /// rate limits) and returns the deserialized `data`.
    pub(crate) async fn query<V: Serialize, D: DeserializeOwned>(
        &self,
        op_name: &'static str,
        document: &'static str,
        variables: V,
    ) -> Result<D> {
        self.run(OpKind::Query, op_name, document, variables).await
    }

    /// Executes one GraphQL **mutation**. Identical to [`Self::query`]
    /// except for retry classification: mutations are never retried on
    /// post-send transport errors / 5xx unless
    /// [`RetryConfig::retry_mutations_on_transient`] is set.
    pub(crate) async fn mutation<V: Serialize, D: DeserializeOwned>(
        &self,
        op_name: &'static str,
        document: &'static str,
        variables: V,
    ) -> Result<D> {
        self.run(OpKind::Mutation, op_name, document, variables)
            .await
    }

    /// Escape hatch: executes an arbitrary GraphQL document and returns the
    /// raw `data` value. Uses the same error classification as typed calls
    /// and is treated as a **query** for retry purposes — do not send
    /// non-idempotent mutations through it unless you can dedupe.
    pub async fn execute_raw(
        &self,
        document: &str,
        variables: serde_json::Value,
    ) -> Result<serde_json::Value> {
        self.execute(OpKind::Query, "execute_raw", document, variables)
            .await
    }

    async fn run<V: Serialize, D: DeserializeOwned>(
        &self,
        kind: OpKind,
        op_name: &'static str,
        document: &str,
        variables: V,
    ) -> Result<D> {
        let variables = serde_json::to_value(variables).map_err(|e| {
            Error::Config(format!("failed to serialize variables for {op_name}: {e}"))
        })?;
        let data = self.execute(kind, op_name, document, variables).await?;
        serde_json::from_value(data).map_err(|source| Error::Decode {
            operation: op_name,
            source,
        })
    }

    async fn execute(
        &self,
        kind: OpKind,
        op_name: &'static str,
        document: &str,
        variables: serde_json::Value,
    ) -> Result<serde_json::Value> {
        let retry = &self.inner.retry;
        let body = serde_json::json!({ "query": document, "variables": variables });
        let auth = self.auth_header()?;
        let max_attempts = retry.max_attempts.max(1);
        let mut attempt: u32 = 0;
        loop {
            attempt += 1;
            let can_retry = attempt < max_attempts;
            let _started = std::time::Instant::now();

            let sent = self
                .inner
                .http
                .post(&self.inner.endpoint)
                .header(AUTHORIZATION, auth.clone())
                .json(&body)
                .send()
                .await;
            let response = match sent {
                Ok(response) => response,
                Err(e) => {
                    // Connect errors mean nothing executed: safe to retry
                    // even mutations. Anything after send may have landed.
                    let retryable = e.is_connect() || self.transient_retry_allowed(kind);
                    if retryable && can_retry {
                        self.backoff(op_name, attempt).await;
                        continue;
                    }
                    return Err(Error::Transport(e));
                }
            };

            let status = response.status();
            let headers = response.headers().clone();
            let info = RateLimitInfo::from_headers(&headers);
            let info = (!info.is_empty()).then_some(info);
            if let Some(info) = &info {
                *self
                    .inner
                    .last_rate_limit
                    .write()
                    .unwrap_or_else(PoisonError::into_inner) = Some(info.clone());
            }

            let bytes = match response.bytes().await {
                Ok(bytes) => bytes,
                Err(e) => {
                    if self.transient_retry_allowed(kind) && can_retry {
                        self.backoff(op_name, attempt).await;
                        continue;
                    }
                    return Err(Error::Transport(e));
                }
            };

            let raw: std::result::Result<RawResponse, serde_json::Error> =
                serde_json::from_slice(&bytes);

            // Rate limited = HTTP 429 OR a GraphQL error with
            // extensions.code == "RATELIMITED" (Linear sends HTTP 400 for
            // this). The server rejected before executing — always safe to
            // retry, mutations included.
            let rate_limited = status == StatusCode::TOO_MANY_REQUESTS
                || raw.as_ref().is_ok_and(|raw| {
                    raw.errors.as_deref().unwrap_or_default().iter().any(|e| {
                        e.extensions.as_ref().and_then(|x| x.code.as_deref()) == Some("RATELIMITED")
                    })
                });
            if rate_limited {
                let wait = rate_limit_wait(&headers, info.as_ref());
                if wait > retry.max_rate_limit_wait || !can_retry {
                    return Err(Error::RateLimited {
                        retry_after: Some(wait),
                        info,
                    });
                }
                #[cfg(feature = "tracing")]
                tracing::warn!(
                    operation = op_name,
                    attempt,
                    wait_ms = wait.as_millis() as u64,
                    "rate limited; waiting for budget reset"
                );
                tokio::time::sleep(wait).await;
                continue;
            }

            if status.is_server_error() {
                if self.transient_retry_allowed(kind) && can_retry {
                    self.backoff(op_name, attempt).await;
                    continue;
                }
                return Err(Error::Http {
                    status: status.as_u16(),
                    body: String::from_utf8_lossy(&bytes).into_owned(),
                });
            }

            let raw = match raw {
                Ok(raw) => raw,
                Err(source) => {
                    if status.is_success() {
                        return Err(Error::Decode {
                            operation: op_name,
                            source,
                        });
                    }
                    return Err(Error::Http {
                        status: status.as_u16(),
                        body: String::from_utf8_lossy(&bytes).into_owned(),
                    });
                }
            };

            // Fail-closed: GraphQL errors surface even when partial data is
            // present.
            if let Some(errors) = raw.errors.filter(|errors| !errors.is_empty()) {
                return Err(Error::Api {
                    operation: op_name,
                    errors,
                });
            }

            if !status.is_success() {
                return Err(Error::Http {
                    status: status.as_u16(),
                    body: String::from_utf8_lossy(&bytes).into_owned(),
                });
            }

            #[cfg(feature = "tracing")]
            tracing::debug!(
                operation = op_name,
                elapsed_ms = _started.elapsed().as_millis() as u64,
                complexity = info.as_ref().and_then(|i| i.complexity_last_query),
                requests_remaining = info.as_ref().and_then(|i| i.requests_remaining),
                "linear-api request"
            );

            return match raw.data {
                Some(data) if !data.is_null() => Ok(data),
                _ => Err(Error::MissingData { operation: op_name }),
            };
        }
    }

    fn transient_retry_allowed(&self, kind: OpKind) -> bool {
        kind == OpKind::Query || self.inner.retry.retry_mutations_on_transient
    }

    fn auth_header(&self) -> Result<HeaderValue> {
        // Linear API keys are sent raw — no "Bearer" prefix.
        let mut value =
            HeaderValue::from_str(self.inner.api_key.expose_secret()).map_err(|_| {
                Error::Config("API key contains characters not permitted in an HTTP header".into())
            })?;
        value.set_sensitive(true);
        Ok(value)
    }

    /// Exponential backoff with full jitter: uniform in
    /// `[0, min(base × 2^(attempt-1), 8s)]`.
    async fn backoff(&self, _op_name: &str, attempt: u32) {
        let exp = self
            .inner
            .retry
            .base_backoff
            .saturating_mul(2u32.saturating_pow(attempt.saturating_sub(1)))
            .min(MAX_BACKOFF);
        let wait = exp.mul_f64(fastrand::f64());
        #[cfg(feature = "tracing")]
        tracing::warn!(
            operation = _op_name,
            attempt,
            backoff_ms = wait.as_millis() as u64,
            "retrying after transient failure"
        );
        tokio::time::sleep(wait).await;
    }
}

/// Wait before retrying a rate-limited request: `Retry-After` seconds when
/// present, else `X-RateLimit-Requests-Reset − now`, floored at 1s.
fn rate_limit_wait(headers: &HeaderMap, info: Option<&RateLimitInfo>) -> Duration {
    let retry_after = headers
        .get(RETRY_AFTER)
        .and_then(|v| v.to_str().ok())
        .and_then(|s| s.trim().parse::<u64>().ok())
        .map(Duration::from_secs);
    let reset_wait = info
        .and_then(|info| info.requests_reset)
        .and_then(|reset| Duration::try_from(reset - time::OffsetDateTime::now_utc()).ok());
    retry_after
        .or(reset_wait)
        .unwrap_or(Duration::ZERO)
        .max(Duration::from_secs(1))
}

#[derive(serde::Deserialize)]
struct RawResponse {
    #[serde(default)]
    data: Option<serde_json::Value>,
    #[serde(default)]
    errors: Option<Vec<GraphQlError>>,
}