perf: speed up stargazer imports

Author: zcesur

lib/algora/admin/admin.ex

@@ -21,7 +21,7 @@ defmodule Algora.Admin do
   alias Algora.Workspace
   alias Algora.Workspace.Installation
   alias Algora.Workspace.Jobs.FetchTopContributions
-  alias Algora.Workspace.Jobs.ImportStargazer
+  alias Algora.Workspace.Jobs.ProcessStargazers
   alias Algora.Workspace.Jobs.SyncUser
   alias Algora.Workspace.Repository
   alias Algora.Workspace.Ticket
@@ -1029,10 +1029,10 @@ defmodule Algora.Admin do
     end
   end
 
-  def import_stargazers(repo_url, max_pages \\ nil) when is_binary(repo_url) do
+  def import_stargazers(repo_url, page \\ 1, max_pages \\ nil) when is_binary(repo_url) do
     with {:ok, [owner: repo_owner, repo: repo_name]} <- parse_repo_url(repo_url),
-         {:ok, repo} <- Workspace.ensure_repository(token(), repo_owner, repo_name) do
-      fetch_and_import_stargazers(repo_owner, repo_name, repo.id, 1, max_pages)
+         {:ok, repo} <- Workspace.ensure_repository(Github.TokenPool.get_token(), repo_owner, repo_name) do
+      fetch_and_import_stargazers(repo_owner, repo_name, repo.id, page, max_pages)
     else
       error ->
         Logger.error("Failed to import stargazers: #{inspect(error)}")
@@ -1047,28 +1047,25 @@ defmodule Algora.Admin do
     else
       url = "/repos/#{repo_owner}/#{repo_name}/stargazers?per_page=100&page=#{page}"
 
-      case Github.Client.fetch(token(), url) do
+      case Github.Client.fetch(Github.TokenPool.get_token(), url) do
         {:ok, []} ->
           Logger.info("Finished importing stargazers - no more pages")
           :ok
 
         {:ok, stargazers} ->
-          stargazers
-          |> Enum.chunk_every(10)
-          |> Enum.each(fn users ->
-            logins = Enum.map(users, fn user -> user["login"] end)
-
-            %{provider_logins: logins, repo_id: repo_id}
-            |> ImportStargazer.new()
-            |> Oban.insert()
-            |> case do
-              {:ok, _job} -> Logger.info("Enqueued job for #{logins}")
-              {:error, error} -> Logger.error("Failed to enqueue job for #{logins}: #{inspect(error)}")
-            end
-          end)
-
-          # Process next page
-          fetch_and_import_stargazers(repo_owner, repo_name, repo_id, page + 1, max_pages)
+          logins = Enum.map(stargazers, fn user -> user["login"] end)
+
+          case %{provider_logins: logins, repo_id: repo_id}
+               |> ProcessStargazers.new()
+               |> Oban.insert() do
+            {:ok, _job} ->
+              Logger.info("Enqueued job for #{logins}")
+              fetch_and_import_stargazers(repo_owner, repo_name, repo_id, page + 1, max_pages)
+
+            {:error, error} ->
+              Logger.error("Failed to enqueue job for #{logins}: #{inspect(error)}")
+              {:error, :failed_to_enqueue_job}
+          end
 
         error ->
           Logger.error("Failed to fetch stargazers page #{page}: #{inspect(error)}")

lib/algora/workspace/jobs/fetch_top_contributions.ex

@@ -2,7 +2,8 @@ defmodule Algora.Workspace.Jobs.FetchTopContributions do
   @moduledoc false
   use Oban.Worker,
     queue: :fetch_top_contributions,
-    max_attempts: 3
+    max_attempts: 2,
+    unique: [period: 30 * 24 * 60 * 60, fields: [:args]]
 
   alias Algora.Github
 

lib/algora/workspace/jobs/import_stargazer.ex

@@ -11,7 +11,7 @@ defmodule Algora.Workspace.Jobs.ImportStargazer do
   @impl Oban.Worker
   def perform(%Oban.Job{args: %{"provider_logins" => provider_logins, "repo_id" => repo_id}}) do
     with {:ok, users} <- Algora.Workspace.fetch_top_contributions_async(Github.TokenPool.get_token(), provider_logins) do
-      {count, _} =
+      {_count, _} =
         Repo.insert_all(
           Stargazer,
           Enum.map(users, fn user ->
@@ -25,13 +25,27 @@ defmodule Algora.Workspace.Jobs.ImportStargazer do
           end)
         )
 
-      if count > 0 do
-        :ok
-      else
-        {:error, :insert_all_failed}
-      end
+      :ok
     end
   end
 
-  def timeout(_), do: :timer.seconds(30)
+  def perform(%Oban.Job{args: %{"provider_login" => provider_login, "repo_id" => repo_id}}) do
+    with {:ok, users} <- Algora.Workspace.fetch_top_contributions_async(Github.TokenPool.get_token(), [provider_login]) do
+      {_count, _} =
+        Repo.insert_all(
+          Stargazer,
+          Enum.map(users, fn user ->
+            %{
+              id: Nanoid.generate(),
+              inserted_at: DateTime.utc_now(),
+              updated_at: DateTime.utc_now(),
+              user_id: user.id,
+              repository_id: repo_id
+            }
+          end)
+        )
+
+      :ok
+    end
+  end
 end

lib/algora/workspace/jobs/process_stargazers.ex

@@ -0,0 +1,54 @@
+defmodule Algora.Workspace.Jobs.ProcessStargazers do
+  @moduledoc false
+  use Oban.Worker,
+    queue: :sync_contribution,
+    max_attempts: 3
+
+  import Ecto.Query
+
+  alias Algora.Github
+  alias Algora.Repo
+  alias Algora.Workspace
+  alias Algora.Workspace.Jobs.ImportStargazer
+  alias Algora.Workspace.Stargazer
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{"provider_logins" => provider_logins, "repo_id" => repo_id}}) do
+    token = Github.TokenPool.get_token()
+
+    jobs =
+      provider_logins
+      |> Enum.reduce([], fn provider_login, acc ->
+        case Workspace.ensure_user(token, provider_login) do
+          {:ok, user} ->
+            if user.provider_meta["followers"] < 10 do
+              Logger.warning("User #{provider_login} has less than 10 followers")
+              acc
+            else
+              if Repo.exists?(from s in Stargazer, where: s.user_id == ^user.id and s.repository_id == ^repo_id) do
+                acc
+              else
+                [user.provider_login | acc]
+              end
+            end
+
+          {:error, reason} ->
+            Logger.error("Failed to fetch user #{provider_login}: #{reason}")
+            acc
+        end
+      end)
+      |> Enum.chunk_every(10)
+      |> Enum.map(fn logins -> ImportStargazer.new(%{provider_logins: logins, repo_id: repo_id}) end)
+      |> Oban.insert_all()
+
+    case jobs do
+      [] ->
+        Logger.warning("No stargazers to process for #{repo_id}")
+
+      _ ->
+        :ok
+    end
+  end
+end

lib/algora/workspace/jobs/sync_contribution.ex

@@ -2,7 +2,7 @@ defmodule Algora.Workspace.Jobs.SyncContribution do
   @moduledoc false
   use Oban.Worker,
     queue: :sync_contribution,
-    max_attempts: 3
+    max_attempts: 2
 
   alias Algora.Github
   alias Algora.Workspace