feat: implement contributor upsert functionality and job processing

- Added `upsert_contributor/3` function to handle contributor data updates or inserts.
- Introduced `ImportContributor` job to fetch and upsert contributor data from GitHub.
- Created `ProcessContributors` job to process contributor data and enqueue import jobs.
- Updated configuration for `sync_contribution` to a new value.

Author: zcesur

config/config.exs

@@ -62,7 +62,7 @@ config :algora, Oban,
     notify_transfer: 1,
     prompt_payout_connect: 1,
     activity_discord: 1,
-    sync_contribution: 1
+    sync_contribution: 20
   ]
 
 # Configures the mailer

lib/algora/workspace/jobs/import_contributor.ex

@@ -0,0 +1,27 @@
+defmodule Algora.Workspace.Jobs.ImportContributor do
+  @moduledoc false
+  use Oban.Worker,
+    queue: :fetch_top_contributions,
+    max_attempts: 3
+
+  alias Algora.Github
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{"contributors_data" => contributors_data, "repo_id" => repo_id}}) do
+    token = Github.TokenPool.get_token()
+    provider_logins = Enum.map(contributors_data, & &1["provider_login"])
+
+    with {:ok, users} <- Algora.Workspace.fetch_top_contributions_async(token, provider_logins) do
+      contributions_map =
+        Map.new(contributors_data, fn contributor -> {contributor["provider_login"], contributor["contributions"]} end)
+
+      for user <- users do
+        contributions = Map.get(contributions_map, user.provider_login, 0)
+        {:ok, _} = Algora.Workspace.upsert_contributor(user.id, repo_id, contributions)
+        {:ok, _} = Algora.Workspace.upsert_user_contribution(user.id, repo_id, contributions)
+      end
+
+      :ok
+    end
+  end
+end

lib/algora/workspace/jobs/process_contributors.ex

@@ -0,0 +1,44 @@
+defmodule Algora.Workspace.Jobs.ProcessContributors do
+  @moduledoc false
+  use Oban.Worker,
+    queue: :sync_contribution,
+    max_attempts: 3
+
+  alias Algora.Github
+  alias Algora.Workspace
+  alias Algora.Workspace.Jobs.ImportContributor
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{"contributors_data" => contributors_data, "repo_id" => repo_id}}) do
+    token = Github.TokenPool.get_token()
+
+    jobs =
+      contributors_data
+      |> Enum.reduce([], fn contributor_data, acc ->
+        provider_login = contributor_data["login"]
+        contributions = contributor_data["contributions"]
+
+        case Workspace.ensure_user(token, provider_login) do
+          {:ok, user} ->
+            [%{provider_login: user.provider_login, contributions: contributions} | acc]
+
+          {:error, reason} ->
+            Logger.error("Failed to fetch user #{provider_login}: #{reason}")
+            acc
+        end
+      end)
+      |> Enum.chunk_every(10)
+      |> Enum.map(fn contributors -> ImportContributor.new(%{contributors_data: contributors, repo_id: repo_id}) end)
+      |> Oban.insert_all()
+
+    case jobs do
+      [] ->
+        Logger.warning("No contributors to process for #{repo_id}")
+
+      _ ->
+        :ok
+    end
+  end
+end

lib/algora/workspace/workspace.ex

@@ -713,6 +713,31 @@ defmodule Algora.Workspace do
     Repo.aggregate(query, :sum, :contribution_count) || 0
   end
 
+  @spec upsert_contributor(String.t(), String.t(), integer()) ::
+          {:ok, Contributor.t()} | {:error, Ecto.Changeset.t()}
+  def upsert_contributor(user_id, repository_id, contribution_count) do
+    attrs = %{
+      user_id: user_id,
+      repository_id: repository_id,
+      contribution_count: contribution_count,
+      last_fetched_at: DateTime.utc_now()
+    }
+
+    case Repo.get_by(Contributor, user_id: user_id, repository_id: repository_id) do
+      nil ->
+        %Contributor{} |> Contributor.changeset(attrs) |> Repo.insert()
+
+      contributor ->
+        if contributor.contributions < contribution_count do
+          contributor
+          |> Contributor.changeset(attrs)
+          |> Repo.update()
+        else
+          {:ok, contributor}
+        end
+    end
+  end
+
   @spec upsert_user_contribution(String.t(), String.t(), integer()) ::
           {:ok, UserContribution.t()} | {:error, Ecto.Changeset.t()}
   def upsert_user_contribution(user_id, repository_id, contribution_count) do
@@ -724,8 +749,17 @@ defmodule Algora.Workspace 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()
+      nil ->
+        %UserContribution{} |> UserContribution.changeset(attrs) |> Repo.insert()
+
+      contribution ->
+        if contribution.contribution_count < contribution_count do
+          contribution
+          |> UserContribution.changeset(attrs)
+          |> Repo.update()
+        else
+          {:ok, contribution}
+        end
     end
   end