<commit_before>defmodule Aelita2.Batcher do
  use GenServer

  alias Aelita2.Repo
  alias Aelita2.Batch
  alias Aelita2.Patch
  alias Aelita2.Status
  alias Aelita2.Integration.GitHub
  alias Aelita2.LinkPatchBatch

  @poll_period 1000

  # Public API

  def start_link do
    GenServer.start_link(__MODULE__, :ok, name: Aelita2.Batcher)
  end

  def reviewed(patch_id) do
    GenServer.cast(Aelita2.Batcher, {:reviewed, patch_id})
  end

  def status(commit, identifier, state, url) do
    GenServer.cast(Aelita2.Batcher, {:status, commit, identifier, state, url})
  end

  # Server callbacks

  def init(:ok) do
    Process.send_after(self(), :poll, @poll_period)
    {:ok, :ok}
  end

  def handle_cast(args, state) do
    Repo.transaction(fn -> do_handle_cast(args) end)
    {:noreply, state}
  end

  def do_handle_cast({:reviewed, patch}) do
    batch = get_new_batch(patch.project_id)
    project_id = batch.project_id
    ^project_id = patch.project_id
    LinkPatchBatch.changeset(%LinkPatchBatch{}, %{batch_id: batch.id, patch_id: patch.id})
    |> Repo.insert!()
  end

  def do_handle_cast({:status, commit, identifier, state, url}) do
    batch = Repo.all(Batch.get_assoc_by_commit(commit))
    case batch do
      [batch] ->
        Status.get_for_batch(batch.id, identifier)
        |> Repo.update_all([set: [state: Status.numberize_state(state), url: url]])
        if batch.state == Batch.numberize_state(:running) do
          maybe_complete_batch(batch)
        end
      [] -> :ok
    end
  end

  def handle_info(:poll, :ok) do
    Repo.transaction(&poll_all/0)
    Process.send_after(self(), :poll, @poll_period)
    {:noreply, :ok}
  end

  # Private implementation details

  defp poll_all() do
    Repo.all(Batch.all_assoc(:incomplete))
    |> Enum.reduce(%{}, &add_batch_to_project_map/2)
    |> Enum.each(&poll_batches/1)
  end

  defp add_batch_to_project_map(batch, project_map) do
    project_id = batch.project_id
    {_, map} = Map.get_and_update(project_map, project_id, &prepend_or_new(&1, batch))
    map
  end

  # This wouldn't be a bad idea for the standard library.
  defp prepend_or_new(list, item) do
    new = if is_nil(list) do
      [item]
    else
      [item | list]
    end
    {item, new}
  end

  defp poll_batches({_project_id, batches}) do
    batches = Enum.sort_by(batches, &{-&1.state, &1.last_polled})
    |> Enum.dedup_by(&(&1.id))
    poll_batches(batches)
  end

  defp poll_batches([%Batch{state: 0} | _] = batches) do
    Enum.each(batches, fn batch -> 0 = batch.state end)
    case Enum.filter(batches, &next_poll_is_past/1) do
      [] -> :ok
      [batch | _] -> start_waiting_batch(batch)
    end
  end

  defp poll_batches([%Batch{state: 1} | _] = batches) do
    Enum.filter(batches, &(&1.state == 1))
    |> Enum.filter(&next_poll_is_past/1)
    |> Enum.each(&poll_running_batch/1)
  end

  defp start_waiting_batch(batch) do
    patches = Repo.all(Patch.all_for_batch(batch.id))
    |> Enum.sort_by(&(&1.pr_xref))
    |> Enum.dedup_by(&(&1.pr_xref))
    if patches != [] do
      start_waiting_batch(batch, patches)
    else
      Repo.delete(batch)
    end
  end
  defp start_waiting_batch(batch, patches) do
    project = batch.project
    token = GitHub.get_installation_token!(project.installation.installation_xref)
    stmp = "#{project.staging_branch}.tmp"
    base = GitHub.copy_branch!(token, project.repo_xref, project.master_branch, stmp)
    do_merge_patch = fn patch, _branch ->
      GitHub.merge_branch!(token, project.repo_xref, patch.commit, stmp, "tmp")
    end
    head = Enum.reduce(patches, base, do_merge_patch)
    parents = [base | Enum.map(patches, &(&1.commit))]
    commit_title = Enum.reduce(patches, "Merge", &"#{&2} \##{&1.pr_xref}")
    commit_body = Enum.reduce(patches, "", &"#{&2}#{&1.title}\n")
    commit_message = "#{commit_title}\n\n#{commit_body}"
    commit = GitHub.synthesize_commit!(token, project.repo_xref, project.staging_branch, head.tree, parents, commit_message)
    setup_statuses(token, project, batch, patches)
    batch = Batch.changeset(batch, %{state: Batch.numberize_state(:running), commit: commit, last_polled: DateTime.to_unix(DateTime.utc_now(), :seconds)})
    |> Repo.update!()
  end

  defp setup_statuses(token, project, batch, patches) do
    toml = GitHub.get_file(token, project.repo_xref, project.staging_branch, "bors.toml")
    case toml do
      nil -> setup_statuses_error(token, project, batch, patches, "bors.toml does not exist")
      toml ->
        case Aelita2.Batcher.BorsToml.new(toml) do
          {:ok, toml} ->
            setup_statuses_ok(token, project, batch, patches, toml)
          {:err, :parse_failed} ->
            setup_statuses_error(token, project, batch, patches, "bors.toml is invalid")
        end
    end
  end

  defp setup_statuses_ok(_token, _project, batch, _patches, toml) do
    Enum.map(toml.status, &%Status{
      batch_id: batch.id,
      identifier: &1,
      url: nil,
      state: Status.numberize_state(:waiting)
      })
    |> Enum.each(&Repo.insert!/1)<commit_msg>Add the PR numbers to the title list<commit_after>defmodule Aelita2.Batcher do
  use GenServer

  alias Aelita2.Repo
  alias Aelita2.Batch
  alias Aelita2.Patch
  alias Aelita2.Status
  alias Aelita2.Integration.GitHub
  alias Aelita2.LinkPatchBatch

  @poll_period 1000

  # Public API

  def start_link do
    GenServer.start_link(__MODULE__, :ok, name: Aelita2.Batcher)
  end

  def reviewed(patch_id) do
    GenServer.cast(Aelita2.Batcher, {:reviewed, patch_id})
  end

  def status(commit, identifier, state, url) do
    GenServer.cast(Aelita2.Batcher, {:status, commit, identifier, state, url})
  end

  # Server callbacks

  def init(:ok) do
    Process.send_after(self(), :poll, @poll_period)
    {:ok, :ok}
  end

  def handle_cast(args, state) do
    Repo.transaction(fn -> do_handle_cast(args) end)
    {:noreply, state}
  end

  def do_handle_cast({:reviewed, patch}) do
    batch = get_new_batch(patch.project_id)
    project_id = batch.project_id
    ^project_id = patch.project_id
    LinkPatchBatch.changeset(%LinkPatchBatch{}, %{batch_id: batch.id, patch_id: patch.id})
    |> Repo.insert!()
  end

  def do_handle_cast({:status, commit, identifier, state, url}) do
    batch = Repo.all(Batch.get_assoc_by_commit(commit))
    case batch do
      [batch] ->
        Status.get_for_batch(batch.id, identifier)
        |> Repo.update_all([set: [state: Status.numberize_state(state), url: url]])
        if batch.state == Batch.numberize_state(:running) do
          maybe_complete_batch(batch)
        end
      [] -> :ok
    end
  end

  def handle_info(:poll, :ok) do
    Repo.transaction(&poll_all/0)
    Process.send_after(self(), :poll, @poll_period)
    {:noreply, :ok}
  end

  # Private implementation details

  defp poll_all() do
    Repo.all(Batch.all_assoc(:incomplete))
    |> Enum.reduce(%{}, &add_batch_to_project_map/2)
    |> Enum.each(&poll_batches/1)
  end

  defp add_batch_to_project_map(batch, project_map) do
    project_id = batch.project_id
    {_, map} = Map.get_and_update(project_map, project_id, &prepend_or_new(&1, batch))
    map
  end

  # This wouldn't be a bad idea for the standard library.
  defp prepend_or_new(list, item) do
    new = if is_nil(list) do
      [item]
    else
      [item | list]
    end
    {item, new}
  end

  defp poll_batches({_project_id, batches}) do
    batches = Enum.sort_by(batches, &{-&1.state, &1.last_polled})
    |> Enum.dedup_by(&(&1.id))
    poll_batches(batches)
  end

  defp poll_batches([%Batch{state: 0} | _] = batches) do
    Enum.each(batches, fn batch -> 0 = batch.state end)
    case Enum.filter(batches, &next_poll_is_past/1) do
      [] -> :ok
      [batch | _] -> start_waiting_batch(batch)
    end
  end

  defp poll_batches([%Batch{state: 1} | _] = batches) do
    Enum.filter(batches, &(&1.state == 1))
    |> Enum.filter(&next_poll_is_past/1)
    |> Enum.each(&poll_running_batch/1)
  end

  defp start_waiting_batch(batch) do
    patches = Repo.all(Patch.all_for_batch(batch.id))
    |> Enum.sort_by(&(&1.pr_xref))
    |> Enum.dedup_by(&(&1.pr_xref))
    if patches != [] do
      start_waiting_batch(batch, patches)
    else
      Repo.delete(batch)
    end
  end
  defp start_waiting_batch(batch, patches) do
    project = batch.project
    token = GitHub.get_installation_token!(project.installation.installation_xref)
    stmp = "#{project.staging_branch}.tmp"
    base = GitHub.copy_branch!(token, project.repo_xref, project.master_branch, stmp)
    do_merge_patch = fn patch, _branch ->
      GitHub.merge_branch!(token, project.repo_xref, patch.commit, stmp, "tmp")
    end
    head = Enum.reduce(patches, base, do_merge_patch)
    parents = [base | Enum.map(patches, &(&1.commit))]
    commit_title = Enum.reduce(patches, "Merge", &"#{&2} \##{&1.pr_xref}")
    commit_body = Enum.reduce(patches, "", &"#{&2}#{&1.pr_xref}: #{&1.title}\n")
    commit_message = "#{commit_title}\n\n#{commit_body}"
    commit = GitHub.synthesize_commit!(token, project.repo_xref, project.staging_branch, head.tree, parents, commit_message)
    setup_statuses(token, project, batch, patches)
    batch = Batch.changeset(batch, %{state: Batch.numberize_state(:running), commit: commit, last_polled: DateTime.to_unix(DateTime.utc_now(), :seconds)})
    |> Repo.update!()
  end

  defp setup_statuses(token, project, batch, patches) do
    toml = GitHub.get_file(token, project.repo_xref, project.staging_branch, "bors.toml")
    case toml do
      nil -> setup_statuses_error(token, project, batch, patches, "bors.toml does not exist")
      toml ->
        case Aelita2.Batcher.BorsToml.new(toml) do
          {:ok, toml} ->
            setup_statuses_ok(token, project, batch, patches, toml)
          {:err, :parse_failed} ->
            setup_statuses_error(token, project, batch, patches, "bors.toml is invalid")
        end
    end
  end

  defp setup_statuses_ok(_token, _project, batch, _patches, toml) do
    Enum.map(toml.status, &%Status{
      batch_id: batch.id,
      identifier: &1,
      url: nil,
      state: Status.numberize_state(:waiting)
      })
    |> Enum.each(&Repo.insert!/1)
{'fraction_non_alphanumeric': 0.13274161735700196, 'fraction_numerical': 0.007790927021696253, 'mean_word_length': 3.4951241134751774, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1269178', 'n_tokens_mistral': 3763, 'n_tokens_neox': 3647, 'n_words': 735}