neiam /gitgud

defmodule GitGud.Repositories.Workers.CommitBackfill do
  @moduledoc """
  Walks `old_sha..new_sha` on a repo and seeds the Postgres git caches.

  Triggered by the on-disk `post-receive` hook for every updated ref.
  Idempotent: rows are SHA-keyed and inserted with `on_conflict: :nothing`.
  """

  use Oban.Worker, queue: :git_cache, max_attempts: 5

  alias GitGud.Federation.Publishing
  alias GitGud.Git
  alias GitGud.Repositories
  alias GitGud.Repositories.Repository
  alias GitGud.Webhooks
  alias GitGud.Workflows

  @batch 100

  @impl Oban.Worker
  def perform(%Oban.Job{
        args: %{
          "repository_id" => rid,
          "ref" => ref,
          "old_sha_hex" => old_hex,
          "new_sha_hex" => new_hex
        } = args
      }) do
    repo = Repositories.get_repository!(rid)
    pusher_id = args["pusher_id"]

    with {:ok, old_sha} <- maybe_hex(old_hex),
         {:ok, new_sha} <- Git.from_hex(new_hex) do
      :ok = Repositories.apply_ref_update(repo, ref, old_sha, new_sha)

      if zero?(new_sha) do
        Webhooks.notify(repo, "push", push_payload(repo, ref, old_sha, new_sha, :delete))
        :ok
      else
        :ok = walk_and_cache(repo, old_sha, new_sha)
        Webhooks.notify(repo, "push", push_payload(repo, ref, old_sha, new_sha, :update))
        dispatch_workflows(repo, ref, new_sha, pusher_id)
        publish_push(repo, ref, old_sha, new_sha)
        refresh_cross_repo_prs(repo, ref, new_sha)
        enqueue_language_stats(repo, ref, new_sha)
        :ok
      end
    end
  end

  defp enqueue_language_stats(repo, "refs/heads/" <> _branch, new_sha) do
    _ =
      GitGud.Repositories.Workers.LanguageStats.new_job(%{
        repository_id: repo.id,
        sha: new_sha
      })
      |> Oban.insert()

    :ok
  end

  defp enqueue_language_stats(_repo, _ref, _sha), do: :ok

  defp refresh_cross_repo_prs(repo, ref, new_sha) do
    branch =
      case ref do
        "refs/heads/" <> b -> b
        _ -> nil
      end

    if branch do
      GitGud.PullRequests.refresh_cross_repo_heads(repo, branch, new_sha)
    end
  end

  defp publish_push(repo, ref, old_sha, new_sha) do
    # Collect a bounded slice of the new commits for the activity payload.
    # We deliberately keep this small — the activity should describe the
    # push, not duplicate the whole history.
    commits =
      case Repositories.list_commits(repo, new_sha, limit: 20) do
        {:ok, list} ->
          list
          |> Enum.map(fn c ->
            %{
              sha: c.sha,
              message: c.message,
              author: c.author_email,
              committed_at: c.committed_at
            }
          end)

        _ ->
          []
      end

    _ =
      Publishing.publish_push(repo, %{
        ref: ref,
        before: sha_hex(old_sha),
        after: sha_hex(new_sha),
        commits: commits
      })

    :ok
  end

  defp push_payload(repo, ref, old_sha, new_sha, kind) do
    %{
      "action" => to_string(kind),
      "ref" => ref,
      "before" => sha_hex(old_sha),
      "after" => sha_hex(new_sha),
      "repository" => %{
        "id" => repo.id,
        "name" => repo.name,
        "default_branch" => repo.default_branch
      }
    }
  end

  defp sha_hex(nil), do: nil
  defp sha_hex(<<0::size(160)>>), do: nil
  defp sha_hex(sha) when is_binary(sha), do: Git.to_hex(sha)

  defp dispatch_workflows(repo, ref, head_sha, pusher_id) do
    trigger =
      cond do
        String.starts_with?(ref, "refs/heads/") -> "push"
        String.starts_with?(ref, "refs/tags/") -> "push"
        true -> nil
      end

    if trigger do
      _ =
        Workflows.dispatch(repo, %{
          trigger: trigger,
          head_sha: head_sha,
          ref: ref,
          triggered_by_id: pusher_id
        })
    end
  end

  defp maybe_hex(hex) when is_binary(hex) and byte_size(hex) == 40 do
    case Git.from_hex(hex) do
      {:ok, sha} -> {:ok, sha}
      :error -> :error
    end
  end

  defp maybe_hex(_), do: {:ok, nil}

  defp zero?(<<0::size(160)>>), do: true
  defp zero?(_), do: false

  defp walk_and_cache(%Repository{} = repo, _old_sha, new_sha) do
    # Page through history newest-first. Stop when we hit a commit we
    # already cached (cheap: indexed lookup on PK).
    case do_walk(repo, new_sha, 0) do
      :ok -> :ok
      {:error, _} -> :ok
    end
  end

  defp do_walk(repo, start_sha, skip) do
    case Repositories.list_commits(repo, start_sha, limit: @batch, skip: skip) do
      {:ok, []} ->
        :ok

      {:ok, commits} ->
        cache_batch(repo, commits)

        if length(commits) < @batch do
          :ok
        else
          do_walk(repo, start_sha, skip + @batch)
        end

      {:error, _} = err ->
        err
    end
  end

  defp cache_batch(%Repository{id: rid, disk_path: path}, commits) do
    Enum.each(commits, fn c ->
      _ = Repositories.upsert_commit(rid, c)

      with {:ok, entries} <- Git.tree(path, c.tree_sha) do
        Repositories.upsert_tree(rid, c.tree_sha, entries)
      end

      base = if c.parent_shas == [], do: nil, else: hd(c.parent_shas)

      with {:ok, stats} <- Git.diff_stats(path, base, c.sha) do
        Repositories.upsert_diffstat(rid, c.sha, stats)
      end
    end)
  end

  @doc """
  Build the Oban job for one ref update. `old_sha`/`new_sha` are
  40-char hex strings (the on-the-wire format from the hook).
  """
  def new_job(%{
        repository_id: rid,
        ref: ref,
        old_sha_hex: old_hex,
        new_sha_hex: new_hex
      } = attrs) do
    __MODULE__.new(%{
      "repository_id" => rid,
      "ref" => ref,
      "old_sha_hex" => old_hex,
      "new_sha_hex" => new_hex,
      "pusher_id" => Map.get(attrs, :pusher_id)
    })
  end
end