Feature/mutex offline user token upsert

CHANGELOG.md

@@ -1,5 +1,7 @@
 ## Unreleased
 
+- Refactor: adds a mutex around the offline shop token and online user token upsert calls.
+
 ## 0.16.4
 
 - Fix: Add support for larger webhook payload bodies (15MB vs. the previous 8MB)

lib/shopify_api/application.ex

@@ -10,7 +10,9 @@ defmodule ShopifyAPI.Application do
     RateLimiting.GraphQLTracker.init()
 
     # Define workers and child supervisors to be supervised
-    children = []
+    children = [
+      {Mutex, name: ShopifyAPI.OfflineToken}
+    ]
 
     # See https://hexdocs.pm/elixir/Supervisor.html
     # for other strategies and supported options

lib/shopify_api/auth_token.ex

@@ -1,4 +1,6 @@
 defmodule ShopifyAPI.AuthToken do
+  require Logger
+
   @derive {Jason.Encoder, only: [:code, :app_name, :shop_name, :token, :timestamp, :plus]}
   defstruct code: "",
             app_name: "",
@@ -32,7 +34,7 @@ defmodule ShopifyAPI.AuthToken do
   def create_key(shop, app), do: "#{shop}:#{app}"
 
   @spec new(App.t(), String.t(), String.t(), String.t()) :: t()
-  def new(app, myshopify_domain, auth_code, token) do
+  def new(%App{} = app, myshopify_domain, auth_code, token) do
     %__MODULE__{
       app_name: app.name,
       shop_name: myshopify_domain,
@@ -42,7 +44,39 @@ defmodule ShopifyAPI.AuthToken do
   end
 
   @spec from_auth_request(App.t(), String.t(), String.t(), map()) :: t()
-  def from_auth_request(app, myshopify_domain, code \\ "", attrs) when is_struct(app, App) do
-    new(app, myshopify_domain, code, attrs["access_token"])
+  def from_auth_request(%App{} = app, myshopify_domain, code \\ "", %{} = attrs),
+    do: new(app, myshopify_domain, code, attrs["access_token"])
+
+  @spec get_offline_token(App.t(), String.t(), String.t()) ::
+          ok_t() | {:error, :failed_fetching_online_token}
+  def get_offline_token(%App{} = app, myshopify_domain, token) do
+    case ShopifyAPI.AuthTokenServer.get(myshopify_domain, app.name) do
+      {:ok, _} = resp -> resp
+      _ -> mutexed_get_offline_token(app, myshopify_domain, token)
+    end
+  end
+
+  defp mutexed_get_offline_token(app, myshopify_domain, token) do
+    mutex_key = {app.name, myshopify_domain}
+
+    Mutex.with_lock(ShopifyAPI.OfflineToken, mutex_key, fn ->
+      # Try fetching a valid token from cache, a new one may have been put in here since the
+      # first call
+      case ShopifyAPI.AuthTokenServer.get(myshopify_domain, app.name) do
+        {:ok, _} = resp -> resp
+        _ -> request_offline_token(app, myshopify_domain, token)
+      end
+    end)
+  end
+
+  defp request_offline_token(app, myshopify_domain, token) do
+    case ShopifyAPI.AuthRequest.request_offline_access_token(app, myshopify_domain, token) do
+      {:ok, token} ->
+        Task.async(fn -> ShopifyAPI.Shop.post_login(token) end)
+        {:ok, token}
+
+      error ->
+        error
+    end
   end
 end

lib/shopify_api/jwt_session_token.ex

@@ -42,68 +42,4 @@ defmodule ShopifyAPI.JWTSessionToken do
 
   def user_id(_),
     do: {:error, "Invalid user token or no id"}
-
-  @spec get_offline_token(JOSE.JWT.t(), String.t()) ::
-          {:ok, ShopifyAPI.AuthToken.t()}
-          | {:error, :invalid_session_token}
-          | {:error, :failed_fetching_online_token}
-  def get_offline_token(%JOSE.JWT{} = jwt, token) do
-    with {:ok, myshopify_domain} <- myshopify_domain(jwt),
-         {:ok, app} <- app(jwt) do
-      case ShopifyAPI.AuthTokenServer.get(myshopify_domain, app.name) do
-        {:ok, _} = resp ->
-          resp
-
-        {:error, _} ->
-          Logger.warning("No token found, exchanging for new")
-
-          case ShopifyAPI.AuthRequest.request_offline_access_token(app, myshopify_domain, token) do
-            {:ok, token} ->
-              fire_post_login_hook(token)
-              {:ok, token}
-
-            error ->
-              error
-          end
-      end
-    else
-      error ->
-        Logger.warning("failed getting required informatio from the JWT #{inspect(error)}")
-        {:error, :invalid_session_token}
-    end
-  end
-
-  @spec get_user_token(JOSE.JWT.t(), String.t()) ::
-          {:ok, ShopifyAPI.UserToken.t()}
-          | {:error, :invalid_session_token}
-          | {:error, :failed_fetching_online_token}
-  def get_user_token(%JOSE.JWT{} = jwt, token) do
-    with {:ok, myshopify_domain} <- myshopify_domain(jwt),
-         {:ok, app} <- app(jwt),
-         {:ok, user_id} <- user_id(jwt) do
-      case ShopifyAPI.UserTokenServer.get_valid(myshopify_domain, app.name, user_id) do
-        {:ok, _} = resp ->
-          resp
-
-        {:error, :invalid_user_token} ->
-          Logger.debug("Expired or no user token found, exchanging for new")
-
-          case ShopifyAPI.AuthRequest.request_online_access_token(app, myshopify_domain, token) do
-            {:ok, user_token} ->
-              fire_post_login_hook(user_token)
-              {:ok, user_token}
-
-            error ->
-              error
-          end
-      end
-    else
-      error ->
-        Logger.warning("failed getting required informatio from the JWT #{inspect(error)}")
-        {:error, :invalid_session_token}
-    end
-  end
-
-  defp fire_post_login_hook(user_token),
-    do: Task.async(fn -> ShopifyAPI.Shop.post_login(user_token) end)
 end

lib/shopify_api/plugs/admin_authenticator.ex

@@ -34,7 +34,9 @@ defmodule ShopifyAPI.Plugs.AdminAuthenticator do
 
   alias Plug.Conn
 
+  alias ShopifyAPI.AuthToken
   alias ShopifyAPI.JWTSessionToken
+  alias ShopifyAPI.UserToken
 
   @defaults [shopify_mount_path: "/shop"]
 
@@ -56,8 +58,8 @@ defmodule ShopifyAPI.Plugs.AdminAuthenticator do
          :ok <- validate_hmac(app, conn.query_params),
          {:ok, myshopify_domain} <- JWTSessionToken.myshopify_domain(jwt),
          {:ok, shop} <- ShopifyAPI.ShopServer.get_or_create(myshopify_domain, true),
-         {:ok, auth_token} <- JWTSessionToken.get_offline_token(jwt, token),
-         {:ok, user_token} <- JWTSessionToken.get_user_token(jwt, token) do
+         {:ok, auth_token} <- get_offline_token(jwt, token),
+         {:ok, user_token} <- get_user_token(jwt, token) do
       conn
       |> assign_app(app)
       |> assign_shop(shop)
@@ -131,4 +133,27 @@ defmodule ShopifyAPI.Plugs.AdminAuthenticator do
       _ -> {:error, :invalid_hmac}
     end)
   end
+
+  defp get_user_token(jwt, token) do
+    with {:ok, app} <- JWTSessionToken.app(jwt),
+         {:ok, myshopify_domain} <- JWTSessionToken.myshopify_domain(jwt),
+         {:ok, user_id} <- JWTSessionToken.user_id(jwt) do
+      UserToken.get_user_token(app, myshopify_domain, user_id, token)
+    else
+      error ->
+        Logger.warning("failed getting required information from the JWT #{inspect(error)}")
+        {:error, :invalid_session_token}
+    end
+  end
+
+  defp get_offline_token(jwt, token) do
+    with {:ok, app} <- JWTSessionToken.app(jwt),
+         {:ok, myshopify_domain} <- JWTSessionToken.myshopify_domain(jwt) do
+      AuthToken.get_offline_token(app, myshopify_domain, token)
+    else
+      error ->
+        Logger.warning("failed getting required information from the JWT #{inspect(error)}")
+        {:error, :invalid_session_token}
+    end
+  end
 end

lib/shopify_api/plugs/auth_shop_session_token.ex

@@ -21,6 +21,7 @@ defmodule ShopifyAPI.Plugs.AuthShopSessionToken do
   alias ShopifyAPI.AuthTokenServer
   alias ShopifyAPI.JWTSessionToken
   alias ShopifyAPI.ShopServer
+  alias ShopifyAPI.UserToken
 
   def init(opts), do: opts
 
@@ -32,7 +33,7 @@ defmodule ShopifyAPI.Plugs.AuthShopSessionToken do
          {:ok, user_id} <- JWTSessionToken.user_id(jwt),
          {:ok, shop} <- ShopServer.get(myshopify_domain),
          {:ok, auth_token} <- AuthTokenServer.get(myshopify_domain, app.name),
-         {:ok, user_token} <- JWTSessionToken.get_user_token(jwt, token) do
+         {:ok, user_token} <- get_user_token(jwt, token) do
       conn
       |> assign(:app, app)
       |> assign(:shop, shop)
@@ -48,4 +49,16 @@ defmodule ShopifyAPI.Plugs.AuthShopSessionToken do
         |> halt()
     end
   end
+
+  defp get_user_token(jwt, token) do
+    with {:ok, app} <- JWTSessionToken.app(jwt),
+         {:ok, myshopify_domain} <- JWTSessionToken.myshopify_domain(jwt),
+         {:ok, user_id} <- JWTSessionToken.user_id(jwt) do
+      UserToken.get_user_token(app, myshopify_domain, user_id, token)
+    else
+      error ->
+        Logger.warning("failed getting required information from the JWT #{inspect(error)}")
+        {:error, :invalid_session_token}
+    end
+  end
 end

lib/shopify_api/user_token.ex

@@ -3,6 +3,7 @@ defmodule ShopifyAPI.UserToken do
   Represents the auth token for individual users, Shopify documentation for the auth process
   is here https://shopify.dev/docs/apps/auth/oauth/getting-started#online-access-mode
   """
+  require Logger
 
   @derive {Jason.Encoder,
            only: [
@@ -79,4 +80,37 @@ defmodule ShopifyAPI.UserToken do
       scope: attrs["scope"]
     )
   end
+
+  @spec get_user_token(App.t(), String.t(), integer(), String.t()) ::
+          {:ok, ShopifyAPI.UserToken.t()} | {:error, :failed_fetching_online_token}
+  def get_user_token(app, myshopify_domain, user_id, token) do
+    case ShopifyAPI.UserTokenServer.get_valid(myshopify_domain, app.name, user_id) do
+      {:ok, _} = resp -> resp
+      _ -> mutexed_get_user_token(app, myshopify_domain, user_id, token)
+    end
+  end
+
+  defp mutexed_get_user_token(app, myshopify_domain, user_id, token) do
+    mutex_key = {app.name, user_id}
+
+    Mutex.with_lock(ShopifyAPI.OfflineToken, mutex_key, fn ->
+      # Try fetching a valid token from cache, a new one may have been put in here since the
+      # get_user_token/4 call
+      case ShopifyAPI.UserTokenServer.get_valid(myshopify_domain, app.name, user_id) do
+        {:ok, _} = resp -> resp
+        _ -> request_offline_token(app, myshopify_domain, token)
+      end
+    end)
+  end
+
+  defp request_offline_token(app, myshopify_domain, token) do
+    case ShopifyAPI.AuthRequest.request_online_access_token(app, myshopify_domain, token) do
+      {:ok, user_token} ->
+        Task.async(fn -> ShopifyAPI.Shop.post_login(user_token) end)
+        {:ok, user_token}
+
+      error ->
+        error
+    end
+  end
 end

mix.exs

@@ -52,6 +52,7 @@ defmodule Plug.ShopifyAPI.MixProject do
       {:httpoison, "~> 2.0"},
       {:jason, "~> 1.0"},
       {:jose, "~> 1.11.2"},
+      {:mutex, "~> 3.0"},
       {:plug, "~> 1.0"},
       {:telemetry, "~> 0.4 or ~> 1.0"}
     ]

mix.lock

@@ -29,6 +29,7 @@
   "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"},
   "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"},
   "mimerl": {:hex, :mimerl, "1.3.0", "d0cd9fc04b9061f82490f6581e0128379830e78535e017f7780f37fea7545726", [:rebar3], [], "hexpm", "a1e15a50d1887217de95f0b9b0793e32853f7c258a5cd227650889b38839fe9d"},
+  "mutex": {:hex, :mutex, "3.0.2", "528877fd0dbc09fc93ad667e10ea0d35a2126fa85205822f9dca85e87d732245", [:mix], [], "hexpm", "0a8f2ed3618160dca6a1e3520b293dc3c2ae53116265e71b4a732d35d29aa3c6"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"},
   "parse_trans": {:hex, :parse_trans, "3.4.1", "6e6aa8167cb44cc8f39441d05193be6e6f4e7c2946cb2759f015f8c56b76e5ff", [:rebar3], [], "hexpm", "620a406ce75dada827b82e453c19cf06776be266f5a67cff34e1ef2cbb60e49a"},
   "plug": {:hex, :plug, "1.18.1", "5067f26f7745b7e31bc3368bc1a2b818b9779faa959b49c934c17730efc911cf", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "57a57db70df2b422b564437d2d33cf8d33cd16339c1edb190cd11b1a3a546cc2"},