neiam /gitgud

defmodule GitGud.Workflows.Concurrency do
  @moduledoc """
  Reading a `concurrency:` block, and serializing the decisions made on it.

  `concurrency` is a group name, or `{group:, cancel-in-progress:}`, either
  value possibly an expression (`GitGud.Workflows.Expr`). Within a group —
  scoped to the repository — at most one member runs and one waits; what
  happens to the others is `GitGud.Workflows`' to decide (`create_run/2`
  for a workflow's group, `enqueue_job/2` for a job's).
  """

  alias GitGud.Repo
  alias GitGud.Workflows.Expr

  # The `concurrency_group` columns are `varchar(255)`.
  @max_group_bytes 255

  @doc """
  Evaluate a `concurrency:` value. `contexts` is a function returning the
  expression contexts, called only when there is something to evaluate.

  Returns `:none` when there's no block, `{:ok, group, cancel_in_progress}`,
  or `{:error, reason}` for one we couldn't read — which the caller treats
  as no concurrency control at all, never as a reason to hold a run back.
  """
  def evaluate(nil, _contexts), do: :none

  def evaluate(%{} = spec, contexts) do
    case Map.fetch(spec, "group") do
      {:ok, group} -> evaluate_group(group, Map.get(spec, "cancel-in-progress"), contexts)
      :error -> {:error, "concurrency has no group"}
    end
  end

  def evaluate(group, contexts) when is_binary(group) or is_number(group),
    do: evaluate_group(group, false, contexts)

  def evaluate(other, _contexts), do: {:error, "unreadable concurrency #{inspect(other)}"}

  defp evaluate_group(group, cancel, contexts) do
    # Contexts cost queries (`vars` decrypts the repository's secrets), so
    # a literal group doesn't build them.
    contexts = if expression?(group) or expression?(cancel), do: contexts.(), else: %{}

    with {:group, {:ok, name}} <- {:group, Expr.interpolate(group, contexts)},
         :ok <- check_name(name),
         {:ok, cancel?} <- cancel_in_progress(cancel, contexts) do
      {:ok, name, cancel?}
    else
      {:group, :error} -> {:error, "could not evaluate group #{inspect(group)}"}
      {:error, _reason} = error -> error
    end
  end

  defp expression?(value), do: is_binary(value) and String.contains?(value, "${{")

  defp check_name(""), do: {:error, "group is empty"}

  defp check_name(name) when byte_size(name) > @max_group_bytes,
    do: {:error, "group is longer than #{@max_group_bytes} bytes"}

  defp check_name(_name), do: :ok

  defp cancel_in_progress(value, _contexts) when value in [nil, false], do: {:ok, false}
  defp cancel_in_progress(true, _contexts), do: {:ok, true}

  defp cancel_in_progress(value, contexts) when is_binary(value) do
    case Expr.interpolate(value, contexts) do
      {:ok, "true"} -> {:ok, true}
      {:ok, flag} when flag in ["false", ""] -> {:ok, false}
      _ -> {:error, "could not evaluate cancel-in-progress #{inspect(value)}"}
    end
  end

  defp cancel_in_progress(value, _contexts),
    do: {:error, "could not evaluate cancel-in-progress #{inspect(value)}"}

  @doc """
  Take the repository's concurrency lock for the rest of the current
  transaction (call it inside one). One lock per repository, for runs' and
  jobs' groups alike: deciding a job's group can cancel a job and finish
  its run, which releases the run's group, and two finer-grained locks
  taken in both orders would deadlock.
  """
  def lock!(repository_id) when is_integer(repository_id) do
    Repo.query!("SELECT pg_advisory_xact_lock($1, $2)", [lock_class(), repository_id])
    :ok
  end

  # The advisory lock's first key: a constant naming this use of it.
  defp lock_class, do: :erlang.phash2(__MODULE__, 2_147_483_647)
end