add oban jobs to fetch top contributions

Author: zcesur

config/config.exs

@@ -57,7 +57,9 @@ config :algora, Oban,
     activity_notifier: 1,
     activity_mailer: 1,
     activity_discord: 10,
-    campaign_emails: 1
+    campaign_emails: 1,
+    fetch_top_contributions: 3,
+    sync_contribution: 3
   ]
 
 # Configures the mailer

lib/algora/admin/admin.ex

@@ -22,29 +22,40 @@ defmodule Algora.Admin do
 
   require Logger
 
-  def add_contributions(handle, opts) do
-    token = token()
-
-    case Workspace.ensure_user(token, handle) do
-      {:ok, user} ->
-        results =
-          Enum.map(opts, fn opts ->
-            with %{repo_name: repo_name, contribution_count: contribution_count} <- opts,
-                 [repo_owner, repo_name] <- String.split(repo_name, "/"),
-                 dbg("fetching repo #{repo_owner}/#{repo_name}"),
-                 {:ok, repo} <- Workspace.ensure_repository(token, repo_owner, repo_name),
-                 {:ok, _tech_stack} <- Workspace.ensure_repo_tech_stack(token, repo),
-                 {:ok, _contribution} <- Algora.Workspace.upsert_user_contribution(user, repo, contribution_count) do
-              :ok
-            end
-          end)
+  def sync_contributions(opts \\ []) do
+    query =
+      User
+      |> where([u], not is_nil(u.handle))
+      |> where([u], not is_nil(u.provider_login))
+      |> where([u], fragment("not exists (select 1 from user_contributions where user_contributions.user_id = ?)", u.id))
 
-        if Enum.any?(results, fn result -> result == :ok end), do: :ok, else: {:error, :failed}
+    query =
+      if handles = opts[:handles] do
+        where(query, [u], u.handle in ^handles)
+      else
+        query
+      end
 
-      {:error, reason} ->
-        Logger.error("Failed to add contributions for #{handle}: #{inspect(reason)}")
-        {:error, reason}
-    end
+    Repo.transaction(
+      fn ->
+        query
+        |> Repo.stream()
+        |> Enum.each(fn user ->
+          if opts[:dry_run] do
+            IO.puts("Enqueued job for #{user.provider_login}")
+          else
+            %{provider_login: user.provider_login}
+            |> Workspace.Jobs.FetchTopContributions.new()
+            |> Oban.insert()
+            |> case do
+              {:ok, _job} -> IO.puts("Enqueued job for #{user.provider_login}")
+              {:error, error} -> IO.puts("Failed to enqueue job for #{user.provider_login}: #{inspect(error)}")
+            end
+          end
+        end)
+      end,
+      timeout: :infinity
+    )
   end
 
   def release_payment(tx_id) do

lib/algora/workspace/jobs/fetch_top_contributions.ex

@@ -0,0 +1,15 @@
+defmodule Algora.Workspace.Jobs.FetchTopContributions do
+  @moduledoc false
+  use Oban.Worker,
+    queue: :fetch_top_contributions,
+    max_attempts: 3,
+    # 30 days
+    unique: [period: 30 * 24 * 60 * 60, fields: [:args]]
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{"provider_login" => provider_login}}) do
+    Algora.Workspace.fetch_top_contributions_async(provider_login)
+  end
+
+  def timeout(_), do: :timer.seconds(10)
+end

lib/algora/workspace/jobs/sync_contribution.ex

@@ -0,0 +1,24 @@
+defmodule Algora.Workspace.Jobs.SyncContribution do
+  @moduledoc false
+  use Oban.Worker,
+    queue: :sync_contribution,
+    max_attempts: 3,
+    # 30 days
+    unique: [period: 30 * 24 * 60 * 60, fields: [:args]]
+
+  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()
+
+    with [repo_owner, repo_name] <- String.split(repo_full_name, "/"),
+         {:ok, repo} <- Workspace.ensure_repository(token, repo_owner, repo_name),
+         {:ok, _tech_stack} <- Workspace.ensure_repo_tech_stack(token, repo),
+         {:ok, _contribution} <- Workspace.upsert_user_contribution(user_id, repo.id, contribution_count) do
+      :ok
+    end
+  end
+end

lib/algora/workspace/workspace.ex

@@ -131,21 +131,13 @@ defmodule Algora.Workspace do
         preload: [user: u]
       )
 
-    res =
-      case Repo.one(repository_query) do
-        %Repository{} = repository -> {:ok, repository}
-        nil -> create_repository_from_github(token, owner, repo)
-      end
-
-    case res do
-      {:ok, repository} -> maybe_schedule_og_image_update(repository)
-      error -> error
+    case Repo.one(repository_query) do
+      %Repository{} = repository -> {:ok, repository}
+      nil -> create_repository_from_github(token, owner, repo)
     end
-
-    res
   end
 
-  defp maybe_schedule_og_image_update(%Repository{} = repository) do
+  def maybe_schedule_og_image_update(%Repository{} = repository) do
     one_day_ago = DateTime.add(DateTime.utc_now(), -1, :day)
 
     needs_update? =
@@ -646,19 +638,84 @@ defmodule Algora.Workspace do
     Repo.all(query)
   end
 
-  @spec upsert_user_contribution(User.t(), Repository.t(), integer()) ::
+  @spec upsert_user_contribution(String.t(), String.t(), integer()) ::
           {:ok, UserContribution.t()} | {:error, Ecto.Changeset.t()}
-  def upsert_user_contribution(%User{} = user, %Repository{} = repository, contribution_count) do
+  def upsert_user_contribution(user_id, repository_id, contribution_count) do
     attrs = %{
-      user_id: user.id,
-      repository_id: repository.id,
+      user_id: user_id,
+      repository_id: repository_id,
       contribution_count: contribution_count,
       last_fetched_at: DateTime.utc_now()
     }
 
-    case Repo.get_by(UserContribution, user_id: user.id, repository_id: repository.id) do
+    case Repo.get_by(UserContribution, user_id: user_id, repository_id: repository_id) do
       nil -> %UserContribution{} |> UserContribution.changeset(attrs) |> Repo.insert()
       contribution -> contribution |> UserContribution.changeset(attrs) |> Repo.update()
     end
   end
+
+  def fetch_top_contributions(provider_login) do
+    token = Algora.Admin.token()
+
+    with {:ok, contributions} <- Algora.Cloud.top_contributions(provider_login),
+         {:ok, user} <- ensure_user(token, provider_login),
+         :ok <- add_contributions(user.id, contributions) do
+      {:ok, contributions}
+    else
+      {:error, reason} ->
+        Logger.error("Failed to fetch contributions for #{provider_login}: #{inspect(reason)}")
+        {:error, reason}
+    end
+  end
+
+  def fetch_top_contributions_async(provider_login) do
+    token = Algora.Admin.token()
+
+    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, nil}
+    else
+      {:error, reason} ->
+        Logger.error("Failed to fetch contributions for #{provider_login}: #{inspect(reason)}")
+        {:error, reason}
+    end
+  end
+
+  def add_contributions_async(user_id, opts) do
+    Repo.transact(fn ->
+      Enum.reduce_while(opts, :ok, fn contribution, _ ->
+        case %{
+               "user_id" => user_id,
+               "repo_full_name" => contribution.repo_name,
+               "contribution_count" => contribution.contribution_count
+             }
+             |> Jobs.SyncContribution.new()
+             |> Oban.insert() do
+          {:ok, _job} -> {:cont, :ok}
+          error -> {:halt, error}
+        end
+      end)
+    end)
+  end
+
+  def add_contributions(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})
+      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()
+
+    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),
+         {:ok, _contribution} <- upsert_user_contribution(user_id, repo.id, contribution_count) do
+      :ok
+    end
+  end
 end

lib/algora_web/live/org/job_live.ex

@@ -729,13 +729,9 @@ defmodule AlgoraWeb.Org.JobLive do
           fn handle ->
             broadcast(socket.assigns.job, {:contributions_fetching, handle})
 
-            with {:ok, contributions} <- Algora.Cloud.top_contributions(handle),
-                 :ok <- Algora.Admin.add_contributions(handle, contributions) do
-              broadcast(socket.assigns.job, {:contributions_fetched, handle, contributions})
-            else
-              {:error, reason} ->
-                Logger.error("Failed to fetch contributions for #{handle}: #{inspect(reason)}")
-                broadcast(socket.assigns.job, {:contributions_failed, handle})
+            case Algora.Workspace.fetch_top_contributions(handle) do
+              {:ok, contributions} -> broadcast(socket.assigns.job, {:contributions_fetched, handle, contributions})
+              {:error, _reason} -> broadcast(socket.assigns.job, {:contributions_failed, handle})
             end
           end,
           timeout: length(users_without_contributions) * 60_000,