use std::borrow::Cow;
use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use autumn_web::actuator::IndicatorGroup;
use autumn_web::app::AppBuilder;
use autumn_web::plugin::Plugin;
use autumn_web::plugin_contract::PluginContract;
use autumn_web::reexports::axum::Router;
use autumn_web::reexports::axum::extract::FromRequestParts;
use autumn_web::reexports::http::request::Parts;
use autumn_web::route_listing::RouteInfo;
use autumn_web::{AppState, AutumnError, AutumnResult};
use crate::client::{RetryPolicy, ShopifyClient};
use crate::config::ShopifyConfig;
use crate::delivery::{DeliveryStore, InMemoryDeliveryStore};
use crate::error::{Error, Result};
use crate::tokens::{InMemoryTokenStore, ShopToken, TokenStore};
use crate::types::{ApiVersion, ShopDomain};
use crate::verify::SecretSet;
use crate::webhooks::{IntoWebhookHandler, WebhookHandler, WebhookRegistry};
const SUPPORTED_AUTUMN_WEB: &str = "0.9";
pub const REFRESH_MARGIN_SECS: i64 = 60;
pub const REFRESH_BACKOFF_SECS: i64 = 10;
type ConfigureFn = Arc<dyn Fn(&mut ShopifyConfig) + Send + Sync + 'static>;
pub type InstallFuture = Pin<Box<dyn Future<Output = AutumnResult<()>> + Send + 'static>>;
pub type InstallHandler = Arc<dyn Fn(InstallContext) -> InstallFuture + Send + Sync + 'static>;
#[derive(Clone)]
#[non_exhaustive]
pub struct InstallContext {
pub shop: ShopDomain,
pub scope: String,
pub state: AppState,
pub shopify: Shopify,
}
impl InstallContext {
#[must_use]
pub fn new(shop: ShopDomain, scope: impl Into<String>, shopify: Shopify) -> Self {
Self {
shop,
scope: scope.into(),
state: shopify.state().clone(),
shopify,
}
}
}
impl std::fmt::Debug for InstallContext {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("InstallContext")
.field("shop", &self.shop)
.field("scope", &self.scope)
.finish_non_exhaustive()
}
}
#[derive(Debug, Default)]
struct RefreshState {
failed_at: Option<i64>,
}
pub(crate) struct Keeper {
http: reqwest::Client,
admin_base_url: Option<String>,
api_key: Option<String>,
secret: String,
tokens: Arc<dyn TokenStore>,
locks: Mutex<HashMap<ShopDomain, Arc<tokio::sync::Mutex<RefreshState>>>>,
}
impl Keeper {
pub(crate) const fn http(&self) -> &reqwest::Client {
&self.http
}
pub(crate) fn api_key(&self) -> Result<&str> {
self.api_key
.as_deref()
.ok_or_else(|| Error::Config("token calls need [shopify] api_key".to_owned()))
}
pub(crate) fn secret(&self) -> &str {
&self.secret
}
pub(crate) fn admin_origin(&self, shop: &ShopDomain) -> String {
self.admin_base_url.as_deref().map_or_else(
|| format!("https://{shop}"),
|base| base.trim_end_matches('/').to_owned(),
)
}
fn lock_for(&self, shop: &ShopDomain) -> Result<Arc<tokio::sync::Mutex<RefreshState>>> {
let mut locks = self.lock_table()?;
Ok(Arc::clone(locks.entry(shop.clone()).or_default()))
}
fn lock_table(
&self,
) -> Result<std::sync::MutexGuard<'_, HashMap<ShopDomain, Arc<tokio::sync::Mutex<RefreshState>>>>>
{
self.locks
.lock()
.map_err(|_| Error::Store("token lock table is poisoned".to_owned()))
}
#[allow(
clippy::significant_drop_tightening,
reason = "the per-shop lock must cover the read, the refresh, and the write"
)]
async fn fresh_token(&self, shop: &ShopDomain, now: i64) -> Result<ShopToken> {
let lock = self.lock_for(shop)?;
let mut state = lock.lock().await;
let token = self
.tokens
.get(shop)
.await?
.ok_or_else(|| Error::NotInstalled(shop.to_string()))?;
if !token.needs_refresh(now, REFRESH_MARGIN_SECS) {
return Ok(token);
}
let backing_off = state
.failed_at
.is_some_and(|at| now.saturating_sub(at) < REFRESH_BACKOFF_SECS);
if backing_off && !token.is_expired(now) {
return Ok(token);
}
let Some(refresh) = token
.refresh_token
.clone()
.filter(|_| token.can_refresh(now))
else {
self.delete_if_unchanged(shop, &token).await?;
return Err(Error::Reauthorize(shop.to_string()));
};
match crate::oauth::refresh_token(self, shop, &refresh, now).await {
Ok(mut fresh) => {
fresh.installed_at = token.installed_at;
self.tokens.put(shop, fresh.clone()).await?;
state.failed_at = None;
tracing::info!(shop = %shop, "shopify token refreshed");
Ok(fresh)
}
Err(Error::Reauthorize(_)) => {
if let Some(current) = self.tokens.get(shop).await?
&& current != token
&& !current.needs_refresh(now, REFRESH_MARGIN_SECS)
{
return Ok(current);
}
self.delete_if_unchanged(shop, &token).await?;
tracing::warn!(shop = %shop, "shopify refused the refresh token. The shop must install the app again");
Err(Error::Reauthorize(shop.to_string()))
}
Err(err) => {
state.failed_at = Some(now);
if token.is_expired(now) {
Err(err)
} else {
tracing::warn!(shop = %shop, error = %err, "shopify refresh failed; the old token is still valid");
Ok(token)
}
}
}
}
async fn delete_if_unchanged(&self, shop: &ShopDomain, seen: &ShopToken) -> Result<()> {
if self.tokens.get(shop).await?.as_ref() == Some(seen) {
self.tokens.delete(shop).await?;
}
Ok(())
}
#[allow(
clippy::significant_drop_tightening,
reason = "the per-shop lock must cover the write"
)]
pub(crate) async fn store_install(&self, shop: &ShopDomain, token: ShopToken) -> Result<()> {
let lock = self.lock_for(shop)?;
let mut state = lock.lock().await;
self.tokens.put(shop, token).await?;
state.failed_at = None;
Ok(())
}
#[allow(
clippy::significant_drop_tightening,
reason = "the per-shop lock must cover the read and the delete"
)]
pub(crate) async fn delete_for_event(
&self,
shop: &ShopDomain,
event_at: Option<i64>,
) -> Result<bool> {
let lock = self.lock_for(shop)?;
let _guard = lock.lock().await;
let current = self.tokens.get(shop).await?;
let newer = matches!(
(current.as_ref().and_then(|t| t.installed_at), event_at),
(Some(installed), Some(event)) if installed > event
);
if newer {
return Ok(false);
}
self.tokens.delete(shop).await?;
Ok(true)
}
}
pub struct ShopifyRuntime {
config: ShopifyConfig,
secrets: SecretSet,
api_version: ApiVersion,
single_store: Option<(ShopDomain, String)>,
retry: RetryPolicy,
registry: WebhookRegistry,
deliveries: Arc<dyn DeliveryStore>,
on_install: Option<InstallHandler>,
keeper: Arc<Keeper>,
}
impl std::fmt::Debug for ShopifyRuntime {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ShopifyRuntime")
.field("config", &self.config)
.field(
"single_store",
&self.single_store.as_ref().map(|(shop, _)| shop),
)
.field("topics", &self.registry.topics())
.finish_non_exhaustive()
}
}
impl ShopifyRuntime {
#[must_use]
pub const fn config(&self) -> &ShopifyConfig {
&self.config
}
pub(crate) const fn secrets(&self) -> &SecretSet {
&self.secrets
}
#[must_use]
pub const fn registry(&self) -> &WebhookRegistry {
&self.registry
}
#[must_use]
pub fn token_store(&self) -> &Arc<dyn TokenStore> {
&self.keeper.tokens
}
#[must_use]
pub const fn delivery_store(&self) -> &Arc<dyn DeliveryStore> {
&self.deliveries
}
pub(crate) const fn on_install(&self) -> Option<&InstallHandler> {
self.on_install.as_ref()
}
pub(crate) fn keeper(&self) -> &Keeper {
&self.keeper
}
pub fn client(&self) -> Result<ShopifyClient> {
let (shop, token) = self.single_store.as_ref().ok_or_else(|| {
Error::Config(format!(
"no single-store client: set [shopify] shop and {}",
crate::config::ACCESS_TOKEN_ENV
))
})?;
Ok(self.make_client(shop, token.clone()))
}
fn make_client(&self, shop: &ShopDomain, token: String) -> ShopifyClient {
ShopifyClient::from_parts(
self.keeper.http.clone(),
shop.clone(),
self.api_version.clone(),
token,
)
.with_base_url(self.keeper.admin_origin(shop))
.with_retry(self.retry)
}
pub async fn client_for(&self, shop: &ShopDomain, now: i64) -> Result<ShopifyClient> {
if let Some((single, token)) = &self.single_store
&& single == shop
{
return Ok(self.make_client(shop, token.clone()));
}
let token = self
.keeper
.tokens
.get(shop)
.await?
.ok_or_else(|| Error::NotInstalled(shop.to_string()))?;
if !token.needs_refresh(now, REFRESH_MARGIN_SECS) {
return Ok(self.make_client(shop, token.access_token));
}
let keeper = Arc::clone(&self.keeper);
let owned = shop.clone();
let token = tokio::spawn(async move { keeper.fresh_token(&owned, now).await })
.await
.map_err(|_| Error::Store("the token refresh task stopped".to_owned()))??;
Ok(self.make_client(shop, token.access_token))
}
}
#[derive(Clone)]
pub struct Shopify {
runtime: Arc<ShopifyRuntime>,
state: AppState,
}
impl std::fmt::Debug for Shopify {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Shopify")
.field("runtime", &self.runtime)
.finish_non_exhaustive()
}
}
impl Shopify {
pub fn from_state(state: &AppState) -> Result<Self> {
let runtime = state
.extension::<ShopifyRuntime>()
.ok_or(Error::MissingRuntime)?;
Ok(Self::from_runtime(runtime, state.clone()))
}
pub(crate) const fn from_runtime(runtime: Arc<ShopifyRuntime>, state: AppState) -> Self {
Self { runtime, state }
}
#[must_use]
pub const fn runtime(&self) -> &Arc<ShopifyRuntime> {
&self.runtime
}
#[must_use]
pub const fn state(&self) -> &AppState {
&self.state
}
pub fn client(&self) -> Result<ShopifyClient> {
self.runtime.client()
}
pub async fn client_for(&self, shop: &ShopDomain) -> Result<ShopifyClient> {
let now = self.state.clock().now().timestamp();
self.runtime.client_for(shop, now).await
}
#[must_use]
pub fn install_url(shop: &ShopDomain) -> String {
format!("{}?shop={shop}", crate::oauth::AUTH_PATH)
}
}
impl FromRequestParts<AppState> for Shopify {
type Rejection = AutumnError;
fn from_request_parts(
_parts: &mut Parts,
state: &AppState,
) -> impl Future<Output = std::result::Result<Self, Self::Rejection>> {
std::future::ready(Self::from_state(state).map_err(|err| err.to_autumn_error()))
}
}
#[derive(Clone, Default)]
pub struct ShopifyPlugin {
configure: Vec<ConfigureFn>,
webhooks: Vec<(String, WebhookHandler)>,
token_store: Option<Arc<dyn TokenStore>>,
delivery_store: Option<Arc<dyn DeliveryStore>>,
admin_base_url: Option<String>,
on_install: Option<InstallHandler>,
}
impl std::fmt::Debug for ShopifyPlugin {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let topics: Vec<&str> = self.webhooks.iter().map(|(t, _)| t.as_str()).collect();
f.debug_struct("ShopifyPlugin")
.field("topics", &topics)
.finish_non_exhaustive()
}
}
impl ShopifyPlugin {
#[must_use]
pub fn new() -> Self {
Self::default()
}
#[must_use]
pub fn configure(
mut self,
configure: impl Fn(&mut ShopifyConfig) + Send + Sync + 'static,
) -> Self {
self.configure.push(Arc::new(configure));
self
}
#[must_use]
pub fn webhook(mut self, topic: impl Into<String>, handler: impl IntoWebhookHandler) -> Self {
self.webhooks.push((topic.into(), handler.into_handler()));
self
}
#[must_use]
pub fn token_store(mut self, store: Arc<dyn TokenStore>) -> Self {
self.token_store = Some(store);
self
}
#[must_use]
pub fn delivery_store(mut self, store: Arc<dyn DeliveryStore>) -> Self {
self.delivery_store = Some(store);
self
}
#[must_use]
pub fn on_install<F, Fut>(mut self, handler: F) -> Self
where
F: Fn(InstallContext) -> Fut + Send + Sync + 'static,
Fut: Future<Output = AutumnResult<()>> + Send + 'static,
{
self.on_install = Some(Arc::new(move |ctx| Box::pin(handler(ctx))));
self
}
#[must_use]
pub fn admin_base_url(mut self, base_url: impl Into<String>) -> Self {
self.admin_base_url = Some(base_url.into());
self
}
pub fn build_runtime(&self, config: ShopifyConfig) -> Result<ShopifyRuntime> {
let mut config = config;
for configure in &self.configure {
configure(&mut config);
}
config.validate()?;
let secrets = config.secrets()?;
let api_version = config.api_version()?;
let single_store = match (config.shop_domain()?, config.access_token()) {
(Some(shop), Some(token)) => Some((shop, token.to_owned())),
_ => None,
};
let mut registry = WebhookRegistry::new();
for (topic, handler) in &self.webhooks {
registry.register(topic, Arc::clone(handler))?;
}
if let Some(base) = &self.admin_base_url {
check_loopback(base)?;
}
let retry = RetryPolicy::new(config.max_throttle_retries, RetryPolicy::default().max_wait);
let keeper = Keeper {
http: crate::client::http_client()?,
admin_base_url: self.admin_base_url.clone(),
api_key: config.api_key.clone(),
secret: secrets.current().to_owned(),
tokens: self
.token_store
.clone()
.unwrap_or_else(|| Arc::new(InMemoryTokenStore::new())),
locks: Mutex::new(HashMap::new()),
};
Ok(ShopifyRuntime {
secrets,
api_version,
single_store,
retry,
registry,
deliveries: self
.delivery_store
.clone()
.unwrap_or_else(|| Arc::new(InMemoryDeliveryStore::default())),
on_install: self.on_install.clone(),
keeper: Arc::new(keeper),
config,
})
}
}
impl ShopifyPlugin {
fn start(&self, state: &AppState, config: ShopifyConfig) -> Result<()> {
let runtime = self.build_runtime(config)?;
for warning in startup_warnings(&runtime, &state.config_arc()) {
tracing::warn!("{warning}");
}
if let Ok(client) = runtime.client() {
let client = client.with_retry(RetryPolicy::new(0, std::time::Duration::ZERO));
let _ = state.health_indicator_registry().register(
"shopify",
IndicatorGroup::HealthOnly,
crate::health::indicator(&client),
);
}
tracing::info!(
topics = runtime.registry().topics().len(),
oauth = runtime.config().oauth_enabled(),
single_store = runtime.single_store.is_some(),
"shopify plugin ready"
);
state.extension_or_insert_with(|| runtime);
Ok(())
}
}
impl Plugin for ShopifyPlugin {
fn name(&self) -> Cow<'static, str> {
Cow::Borrowed("autumn-plugin-shopify")
}
fn contract(&self) -> Option<PluginContract> {
Some(
PluginContract::new(env!("CARGO_PKG_NAME"))
.plugin_version(env!("CARGO_PKG_VERSION"))
.autumn_web(SUPPORTED_AUTUMN_WEB),
)
}
fn build(self, app: AppBuilder) -> AppBuilder {
let (router, infos) = mount();
app.config_section(crate::config::CONFIG_SECTION)
.on_startup(move |state: AppState| {
let plugin = self.clone();
async move {
let config = ShopifyConfig::load().map_err(|err| err.to_autumn_error())?;
plugin
.start(&state, config)
.map_err(|err| err.to_autumn_error())
}
})
.nest("/shopify", router)
.declare_plugin_routes(infos)
}
}
pub(crate) fn startup_warnings(
runtime: &ShopifyRuntime,
app_config: &autumn_web::config::AutumnConfig,
) -> Vec<String> {
let mut warnings = Vec::new();
if runtime.config().oauth_enabled() {
for topic in crate::types::Topic::COMPLIANCE {
if !runtime.registry().contains(topic) {
warnings.push(format!(
"shopify: no handler for the mandatory compliance topic {topic}; \
public apps must handle it"
));
}
}
}
let config = runtime.config();
if config.shop.is_some() != config.access_token().is_some() {
warnings.push(format!(
"shopify: single-store mode is off; it needs both [shopify] shop and {}",
crate::config::ACCESS_TOKEN_ENV
));
}
let csrf = &app_config.security.csrf;
if csrf.enabled
&& !csrf
.exempt_paths
.iter()
.any(|path| covers(path, WEBHOOK_PATH))
{
warnings.push(format!(
"shopify: CSRF checks block {WEBHOOK_PATH}; add it to security.csrf.exempt_paths"
));
}
warnings
}
pub(crate) fn mount() -> (Router<AppState>, Vec<RouteInfo>) {
let all_routes: Vec<autumn_web::Route> = crate::webhooks::webhook_routes()
.into_iter()
.chain(crate::oauth::oauth_routes())
.collect();
let mut infos = autumn_web::route_listing::collect_route_infos(&all_routes, &[], &[], &[])
.unwrap_or_default();
for info in &mut infos {
info.path = format!("/shopify{}", info.path);
}
let mut router = Router::<AppState>::new();
for route in all_routes {
router = router.route(route.path, route.handler);
}
(router, infos)
}
fn check_loopback(base: &str) -> Result<()> {
let url = url::Url::parse(base)
.map_err(|_| Error::Config(format!("admin_base_url {base:?} is not a valid URL")))?;
let loopback = matches!(url.host_str(), Some("localhost" | "127.0.0.1" | "[::1]"));
if loopback && matches!(url.scheme(), "http" | "https") {
Ok(())
} else {
Err(Error::Config(
"admin_base_url is for tests only and must use a loopback host".to_owned(),
))
}
}
const WEBHOOK_PATH: &str = "/shopify/webhooks";
fn covers(prefix: &str, path: &str) -> bool {
path == prefix
|| (path.starts_with(prefix)
&& (prefix.ends_with('/') || path.as_bytes().get(prefix.len()) == Some(&b'/')))
}
#[cfg(test)]
mod tests;