grab tokens from tokenpool while fetching top contributions

Author: zcesur

config/config.exs

@@ -59,7 +59,7 @@ config :algora, Oban,
     activity_discord: 10,
     campaign_emails: 1,
     fetch_top_contributions: 1,
-    sync_contribution: 3
+    sync_contribution: 20
   ]
 
 # Configures the mailer

lib/algora/integrations/github/token_pool.ex

@@ -15,7 +15,14 @@ defmodule Algora.Github.TokenPool do
   end
 
   def get_token do
-    GenServer.call(__MODULE__, :get_token)
+    case maybe_get_token() do
+      token when is_binary(token) -> token
+      _ -> get_token()
+    end
+  end
+
+  def maybe_get_token do
+    GenServer.call(__MODULE__, :maybe_get_token)
   end
 
   def refresh_tokens do
@@ -35,7 +42,7 @@ defmodule Algora.Github.TokenPool do
   end
 
   @impl true
-  def handle_call(:get_token, _from, %{current_token_index: index, tokens: tokens} = state) do
+  def handle_call(:maybe_get_token, _from, %{current_token_index: index, tokens: tokens} = state) do
     token = Enum.at(tokens, index)
 
     if token == nil do
@@ -44,7 +51,16 @@ defmodule Algora.Github.TokenPool do
       next_index = rem(index + 1, length(tokens))
       if next_index == 0, do: refresh_tokens()
 
-      {:reply, token, %{state | current_token_index: next_index}}
+      case Github.get_current_user(token) do
+        {:ok, _} ->
+          dbg(token)
+          dbg("token is valid")
+          {:reply, token, %{state | current_token_index: next_index}}
+
+        _ ->
+          dbg("token is invalid")
+          {:reply, nil, %{state | current_token_index: next_index}}
+      end
     end
   end
 

lib/algora/workspace/jobs/fetch_top_contributions.ex

@@ -6,9 +6,11 @@ defmodule Algora.Workspace.Jobs.FetchTopContributions do
     # 30 days
     unique: [period: 30 * 24 * 60 * 60, fields: [:args]]
 
+  alias Algora.Github
+
   @impl Oban.Worker
   def perform(%Oban.Job{args: %{"provider_login" => provider_login}}) do
-    Algora.Workspace.fetch_top_contributions_async(provider_login)
+    Algora.Workspace.fetch_top_contributions_async(Github.TokenPool.get_token(), provider_login)
   end
 
   def timeout(_), do: :timer.seconds(30)

lib/algora/workspace/jobs/sync_contribution.ex

@@ -6,13 +6,14 @@ defmodule Algora.Workspace.Jobs.SyncContribution do
     # 30 days
     unique: [period: 30 * 24 * 60 * 60, fields: [:args]]
 
+  alias Algora.Github
   alias Algora.Workspace
 
   @impl Oban.Worker
   def perform(%Oban.Job{
         args: %{"user_id" => user_id, "repo_full_name" => repo_full_name, "contribution_count" => contribution_count}
       }) do
-    token = Algora.Admin.token()
+    token = Github.TokenPool.get_token()
 
     with [repo_owner, repo_name] <- String.split(repo_full_name, "/"),
          {:ok, repo} <- Workspace.ensure_repository(token, repo_owner, repo_name),

lib/algora/workspace/workspace.ex

@@ -654,12 +654,10 @@ defmodule Algora.Workspace do
     end
   end
 
-  def fetch_top_contributions(provider_login) do
-    token = Algora.Admin.token()
-
+  def fetch_top_contributions(token, provider_login) do
     with {:ok, contributions} <- Algora.Cloud.top_contributions(provider_login),
          {:ok, user} <- ensure_user(token, provider_login),
-         :ok <- add_contributions(user.id, contributions) do
+         :ok <- add_contributions(token, user.id, contributions) do
       {:ok, contributions}
     else
       {:error, reason} ->
@@ -668,12 +666,10 @@ defmodule Algora.Workspace do
     end
   end
 
-  def fetch_top_contributions_async(provider_login) do
-    token = Algora.Admin.token()
-
+  def fetch_top_contributions_async(token, provider_login) do
     with {:ok, contributions} <- Algora.Cloud.top_contributions(provider_login),
          {:ok, user} <- ensure_user(token, provider_login),
-         {:ok, _} <- add_contributions_async(user.id, contributions) do
+         {:ok, _} <- add_contributions_async(token, user.id, contributions) do
       {:ok, nil}
     else
       {:error, reason} ->
@@ -682,7 +678,7 @@ defmodule Algora.Workspace do
     end
   end
 
-  def add_contributions_async(user_id, opts) do
+  def add_contributions_async(_token, user_id, opts) do
     Repo.transact(fn ->
       Enum.reduce_while(opts, :ok, fn contribution, _ ->
         case %{
@@ -699,18 +695,26 @@ defmodule Algora.Workspace do
     end)
   end
 
-  def add_contributions(user_id, opts) do
+  def add_contributions(token, user_id, opts) do
     results =
       Enum.map(opts, fn %{repo_name: repo_name, contribution_count: contribution_count} ->
-        add_contribution(%{user_id: user_id, repo_full_name: repo_name, contribution_count: contribution_count})
+        add_contribution(%{
+          token: token,
+          user_id: user_id,
+          repo_full_name: repo_name,
+          contribution_count: contribution_count
+        })
       end)
 
     if Enum.any?(results, fn result -> result == :ok end), do: :ok, else: {:error, :failed}
   end
 
-  def add_contribution(%{user_id: user_id, repo_full_name: repo_full_name, contribution_count: contribution_count}) do
-    token = Algora.Admin.token()
-
+  def add_contribution(%{
+        token: token,
+        user_id: user_id,
+        repo_full_name: repo_full_name,
+        contribution_count: contribution_count
+      }) do
     with [repo_owner, repo_name] <- String.split(repo_full_name, "/"),
          {:ok, repo} <- ensure_repository(token, repo_owner, repo_name),
          {:ok, _tech_stack} <- ensure_repo_tech_stack(token, repo),

lib/algora_web/live/org/job_live.ex

@@ -731,7 +731,7 @@ defmodule AlgoraWeb.Org.JobLive do
           fn handle ->
             broadcast(socket.assigns.job, {:contributions_fetching, handle})
 
-            case Algora.Workspace.fetch_top_contributions(handle) do
+            case Algora.Workspace.fetch_top_contributions(Algora.Admin.token(), handle) do
               {:ok, contributions} -> broadcast(socket.assigns.job, {:contributions_fetched, handle, contributions})
               {:error, _reason} -> broadcast(socket.assigns.job, {:contributions_failed, handle})
             end