autumn_plugin_shopify/
client.rs

1//! Admin GraphQL API client.
2//!
3//! [`ShopifyClient`] posts GraphQL documents to
4//! `https://{shop}/admin/api/{version}/graphql.json` with the
5//! `X-Shopify-Access-Token` header.
6//!
7//! Shopify does not run a throttled request, so a retry is safe. The
8//! client retries a `THROTTLED` GraphQL error and an HTTP 429, up to
9//! [`RetryPolicy::max_retries`]. It does not retry other failures: a
10//! second run of a mutation can change data two times.
11
12use std::time::Duration;
13
14use serde::Deserialize;
15use serde::de::DeserializeOwned;
16
17use crate::error::{Error, Result};
18use crate::types::{ApiVersion, ShopDomain, Topic};
19
20/// Request timeout for Shopify calls.
21pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
22/// The longest error body the client keeps, in chars.
23pub(crate) const MAX_ERROR_BODY_CHARS: usize = 500;
24/// Header that carries the Admin API access token.
25pub const ACCESS_TOKEN_HEADER: &str = "x-shopify-access-token";
26
27/// How the client retries throttled calls.
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29#[non_exhaustive]
30pub struct RetryPolicy {
31    /// The number of retries after the first try.
32    pub max_retries: u32,
33    /// The longest wait before one retry.
34    pub max_wait: Duration,
35}
36
37impl RetryPolicy {
38    /// Make a policy.
39    #[must_use]
40    pub const fn new(max_retries: u32, max_wait: Duration) -> Self {
41        Self {
42            max_retries,
43            max_wait,
44        }
45    }
46}
47
48impl Default for RetryPolicy {
49    fn default() -> Self {
50        Self {
51            max_retries: 3,
52            max_wait: Duration::from_secs(10),
53        }
54    }
55}
56
57/// The shop record from [`ShopifyClient::shop_info`].
58#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
59#[serde(rename_all = "camelCase")]
60#[non_exhaustive]
61pub struct ShopInfo {
62    /// The shop name.
63    pub name: String,
64    /// The `*.myshopify.com` domain.
65    pub myshopify_domain: String,
66}
67
68/// An Admin GraphQL client for one shop and one token.
69///
70/// A clone is cheap. `Debug` redacts the token.
71#[derive(Clone)]
72pub struct ShopifyClient {
73    http: reqwest::Client,
74    shop: ShopDomain,
75    base_url: String,
76    api_version: ApiVersion,
77    token: String,
78    retry: RetryPolicy,
79}
80
81impl std::fmt::Debug for ShopifyClient {
82    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
83        f.debug_struct("ShopifyClient")
84            .field("shop", &self.shop)
85            .field("base_url", &self.base_url)
86            .field("api_version", &self.api_version)
87            .field("token", &"<redacted>")
88            .field("retry", &self.retry)
89            .finish_non_exhaustive()
90    }
91}
92
93impl ShopifyClient {
94    /// Make a client for `shop` with a new HTTP connection pool.
95    ///
96    /// # Errors
97    ///
98    /// Returns [`Error::Config`] when the token is blank, or
99    /// [`Error::Http`] when the HTTP client does not build.
100    pub fn new(
101        shop: ShopDomain,
102        api_version: ApiVersion,
103        access_token: impl Into<String>,
104    ) -> Result<Self> {
105        let token = access_token.into();
106        if token.trim().is_empty() {
107            return Err(Error::Config("the access token is blank".to_owned()));
108        }
109        Ok(Self::from_parts(http_client()?, shop, api_version, token))
110    }
111
112    /// Make a client on a shared HTTP pool.
113    pub(crate) fn from_parts(
114        http: reqwest::Client,
115        shop: ShopDomain,
116        api_version: ApiVersion,
117        token: String,
118    ) -> Self {
119        let base_url = format!("https://{shop}");
120        Self {
121            http,
122            shop,
123            base_url,
124            api_version,
125            token,
126            retry: RetryPolicy::default(),
127        }
128    }
129
130    /// Send requests to `base_url` instead of `https://{shop}`. Use it for
131    /// a local stub in tests. The requests carry the access token, so use
132    /// only a host you trust.
133    #[must_use]
134    pub fn with_base_url(mut self, base_url: impl Into<String>) -> Self {
135        base_url
136            .into()
137            .trim_end_matches('/')
138            .clone_into(&mut self.base_url);
139        self
140    }
141
142    /// Set the retry policy.
143    #[must_use]
144    pub const fn with_retry(mut self, retry: RetryPolicy) -> Self {
145        self.retry = retry;
146        self
147    }
148
149    /// The shop this client calls.
150    #[must_use]
151    pub const fn shop(&self) -> &ShopDomain {
152        &self.shop
153    }
154
155    /// The API version this client uses.
156    #[must_use]
157    pub const fn api_version(&self) -> &ApiVersion {
158        &self.api_version
159    }
160
161    /// The GraphQL endpoint URL.
162    #[must_use]
163    pub fn endpoint(&self) -> String {
164        format!(
165            "{}/admin/api/{}/graphql.json",
166            self.base_url, self.api_version
167        )
168    }
169
170    /// Run a GraphQL document and return its `data`.
171    ///
172    /// # Errors
173    ///
174    /// Returns [`Error::Http`] on a transport failure,
175    /// [`Error::ShopifyApi`] on a non-success status, [`Error::GraphQl`]
176    /// when the response has errors, [`Error::UnexpectedResponse`] when the
177    /// body has no `data`, and [`Error::Throttled`] when the retries run
178    /// out.
179    pub async fn graphql(
180        &self,
181        query: &str,
182        variables: serde_json::Value,
183    ) -> Result<serde_json::Value> {
184        let payload = serde_json::json!({ "query": query, "variables": variables });
185        let mut retries = 0;
186        loop {
187            let response = self
188                .http
189                .post(self.endpoint())
190                .header(ACCESS_TOKEN_HEADER, &self.token)
191                .json(&payload)
192                .send()
193                .await
194                .map_err(Error::Http)?;
195            let wait = if response.status() == reqwest::StatusCode::TOO_MANY_REQUESTS {
196                retry_after(response.headers())
197            } else if response.status().is_success() {
198                let body: serde_json::Value = read_json(response).await?;
199                match throttle_wait(&body) {
200                    Some(wait) => wait,
201                    None => return into_data(body),
202                }
203            } else {
204                return Err(api_error(response).await);
205            };
206            if retries >= self.retry.max_retries {
207                return Err(Error::Throttled { retries });
208            }
209            retries += 1;
210            tracing::debug!(shop = %self.shop, retries, "shopify throttled the call; retrying");
211            tokio::time::sleep(wait.min(self.retry.max_wait)).await;
212        }
213    }
214
215    /// Run a GraphQL document and deserialize its `data` into `T`.
216    ///
217    /// # Errors
218    ///
219    /// See [`graphql`](Self::graphql). Also returns
220    /// [`Error::UnexpectedResponse`] when `data` does not match `T`.
221    pub async fn query<T: DeserializeOwned>(
222        &self,
223        query: &str,
224        variables: serde_json::Value,
225    ) -> Result<T> {
226        let data = self.graphql(query, variables).await?;
227        serde_json::from_value(data)
228            .map_err(|err| unexpected(&format!("data does not match: {err}")))
229    }
230
231    /// Read the shop name and domain. The health check uses it.
232    ///
233    /// # Errors
234    ///
235    /// See [`query`](Self::query).
236    pub async fn shop_info(&self) -> Result<ShopInfo> {
237        #[derive(Deserialize)]
238        struct Data {
239            shop: ShopInfo,
240        }
241        let data: Data = self
242            .query(
243                "query AutumnShopInfo { shop { name myshopifyDomain } }",
244                serde_json::json!({}),
245            )
246            .await?;
247        Ok(data.shop)
248    }
249
250    /// Subscribe `callback_url` to `topic` (for example `orders/create`)
251    /// with `webhookSubscriptionCreate`. Returns the subscription id.
252    ///
253    /// The compliance topics are not in the GraphQL topic enum. Declare
254    /// them in `shopify.app.toml` (`compliance_topics`).
255    ///
256    /// # Errors
257    ///
258    /// Returns [`Error::InvalidRequest`] for a malformed topic and
259    /// [`Error::Config`] for a compliance topic. See also
260    /// [`query`](Self::query). Returns [`Error::GraphQl`] with the
261    /// `userErrors` messages when Shopify refuses the subscription.
262    pub async fn subscribe_webhook(&self, topic: &str, callback_url: &str) -> Result<String> {
263        #[derive(Deserialize)]
264        #[serde(rename_all = "camelCase")]
265        struct Data {
266            webhook_subscription_create: Payload,
267        }
268        #[derive(Deserialize)]
269        #[serde(rename_all = "camelCase")]
270        struct Payload {
271            webhook_subscription: Option<Subscription>,
272            #[serde(default)]
273            user_errors: Vec<UserError>,
274        }
275        #[derive(Deserialize)]
276        struct Subscription {
277            id: String,
278        }
279        #[derive(Deserialize)]
280        struct UserError {
281            message: String,
282        }
283        let topic = Topic::parse(topic)?;
284        if topic.is_compliance() {
285            return Err(Error::Config(format!(
286                "{topic} is a compliance topic; declare it in shopify.app.toml compliance_topics"
287            )));
288        }
289        let data: Data = self
290            .query(
291                SUBSCRIBE_MUTATION,
292                serde_json::json!({
293                    "topic": topic.graphql_enum(),
294                    "webhookSubscription": { "uri": callback_url },
295                }),
296            )
297            .await?;
298        let payload = data.webhook_subscription_create;
299        if !payload.user_errors.is_empty() {
300            return Err(Error::GraphQl(
301                payload.user_errors.into_iter().map(|e| e.message).collect(),
302            ));
303        }
304        payload
305            .webhook_subscription
306            .map(|subscription| subscription.id)
307            .ok_or_else(|| unexpected("webhookSubscriptionCreate returned no subscription"))
308    }
309}
310
311/// Build the shared HTTP client: timeout and user agent.
312///
313/// # Errors
314///
315/// Returns [`Error::Http`] when the client does not build.
316pub(crate) fn http_client() -> Result<reqwest::Client> {
317    // No redirects: a redirect would carry the access token or the client
318    // secret to another host.
319    reqwest::Client::builder()
320        .timeout(REQUEST_TIMEOUT)
321        .redirect(reqwest::redirect::Policy::none())
322        .user_agent(concat!("autumn-plugin-shopify/", env!("CARGO_PKG_VERSION")))
323        .build()
324        .map_err(Error::Http)
325}
326
327/// Turn a non-success response into [`Error::ShopifyApi`] with a
328/// truncated body.
329pub(crate) async fn api_error(response: reqwest::Response) -> Error {
330    let status = response.status().as_u16();
331    let body = response.text().await.unwrap_or_default();
332    Error::ShopifyApi {
333        status,
334        body: body.chars().take(MAX_ERROR_BODY_CHARS).collect(),
335    }
336}
337
338/// Read a success body as JSON.
339async fn read_json(response: reqwest::Response) -> Result<serde_json::Value> {
340    let bytes = response.bytes().await.map_err(Error::Http)?;
341    serde_json::from_slice(&bytes).map_err(|_| unexpected("the response is not JSON"))
342}
343
344/// Return `data`, or the `errors` messages.
345fn into_data(mut body: serde_json::Value) -> Result<serde_json::Value> {
346    if let Some(errors) = body.get("errors").and_then(serde_json::Value::as_array)
347        && !errors.is_empty()
348    {
349        let messages = errors
350            .iter()
351            .map(|error| {
352                error
353                    .get("message")
354                    .and_then(serde_json::Value::as_str)
355                    .unwrap_or("unknown error")
356                    .to_owned()
357            })
358            .collect();
359        return Err(Error::GraphQl(messages));
360    }
361    match body.get_mut("data").map(serde_json::Value::take) {
362        Some(data) if !data.is_null() => Ok(data),
363        _ => Err(unexpected("the response has no data")),
364    }
365}
366
367/// An [`Error::UnexpectedResponse`] for a 200 answer of the wrong shape.
368fn unexpected(detail: &str) -> Error {
369    Error::UnexpectedResponse(detail.chars().take(MAX_ERROR_BODY_CHARS).collect())
370}
371
372/// The wait a `THROTTLED` response asks for, from `extensions.cost`.
373///
374/// Wait = (requested cost - available points) / restore rate, at least
375/// [`MIN_THROTTLE_WAIT`]. One second when the cost data is missing.
376/// `None` when the response is not throttled.
377pub(crate) fn throttle_wait(body: &serde_json::Value) -> Option<Duration> {
378    let throttled = body.get("errors")?.as_array()?.iter().any(|error| {
379        error
380            .pointer("/extensions/code")
381            .and_then(serde_json::Value::as_str)
382            == Some("THROTTLED")
383    });
384    if !throttled {
385        return None;
386    }
387    let cost = body.pointer("/extensions/cost");
388    let number = |path: &str| {
389        cost.and_then(|c| c.pointer(path))
390            .and_then(serde_json::Value::as_f64)
391    };
392    let wait = match (
393        number("/requestedQueryCost"),
394        number("/throttleStatus/currentlyAvailable"),
395        number("/throttleStatus/restoreRate"),
396    ) {
397        (Some(requested), Some(available), Some(rate)) if rate > 0.0 => {
398            Duration::try_from_secs_f64(((requested - available) / rate).max(0.0))
399                .unwrap_or(DEFAULT_WAIT)
400                .max(MIN_THROTTLE_WAIT)
401        }
402        _ => DEFAULT_WAIT,
403    };
404    Some(wait)
405}
406
407/// The wait an HTTP 429 asks for, from `Retry-After` (seconds).
408/// One second when the header is missing or bad.
409pub(crate) fn retry_after(headers: &reqwest::header::HeaderMap) -> Duration {
410    headers
411        .get(reqwest::header::RETRY_AFTER)
412        .and_then(|value| value.to_str().ok())
413        .and_then(|value| value.trim().parse::<f64>().ok())
414        .and_then(|secs| Duration::try_from_secs_f64(secs).ok())
415        .unwrap_or(DEFAULT_WAIT)
416}
417
418/// The wait when Shopify gives no hint. Shopify recommends one second.
419const DEFAULT_WAIT: Duration = Duration::from_secs(1);
420/// The shortest wait after a throttle.
421const MIN_THROTTLE_WAIT: Duration = Duration::from_millis(100);
422
423/// The `webhookSubscriptionCreate` mutation.
424const SUBSCRIBE_MUTATION: &str = "mutation AutumnSubscribeWebhook($topic: WebhookSubscriptionTopic!, $webhookSubscription: WebhookSubscriptionInput!) { webhookSubscriptionCreate(topic: $topic, webhookSubscription: $webhookSubscription) { webhookSubscription { id } userErrors { field message } } }";
425
426#[cfg(test)]
427mod tests;
autumn_plugin_shopify/client.rs

autumn_plugin_shopify/
client.rs
