viewstamped_replication

defmodule ViewstampedReplication do
  @moduledoc """
  Protocol-generic Viewstamped Replication runtime.
  """

  alias ViewstampedReplication.{Client, Configuration, Replica, ReplicaSupervisor}

  @spec start_replica(keyword()) :: {:ok, pid()} | {:error, term()}
  def start_replica(opts) do
    with {:ok, configuration} <- configuration(opts),
         child_opts <- Keyword.put(opts, :configuration, configuration),
         {:ok, _supervisor} <-
           DynamicSupervisor.start_child(
             ViewstampedReplication.ReplicaDynamicSupervisor,
             {ReplicaSupervisor, child_opts}
           ),
         pid when is_pid(pid) <- Replica.whereis(configuration.group_id, configuration.replica_id) do
      {:ok, pid}
    else
      {:error, {:already_started, _supervisor}} -> {:error, :already_started}
      {:error, _reason} = error -> error
      nil -> {:error, :replica_start_failed}
    end
  end

  @spec stop_replica(term(), term()) :: :ok | {:error, :not_found}
  def stop_replica(group_id, replica_id) do
    case ReplicaSupervisor.whereis(group_id, replica_id) do
      nil ->
        {:error, :not_found}

      supervisor ->
        DynamicSupervisor.terminate_child(
          ViewstampedReplication.ReplicaDynamicSupervisor,
          supervisor
        )
    end
  end

  @spec status(term(), term()) :: {:ok, map()} | {:error, :not_found}
  def status(group_id, replica_id), do: Replica.status(group_id, replica_id)

  @spec primary(term(), term()) :: {:ok, term()} | {:error, :not_found}
  def primary(group_id, replica_id), do: Replica.primary(group_id, replica_id)

  @spec snapshot(term(), term()) :: :ok | {:error, term()}
  def snapshot(group_id, replica_id), do: Replica.snapshot(group_id, replica_id)

  @spec command(term(), term(), keyword()) :: {:ok, term()} | {:error, term()}
  def command(group_id, operation, opts) do
    with {:ok, client} <- Keyword.fetch(opts, :client) do
      Client.command(
        client,
        operation,
        group_id: group_id,
        timeout: Keyword.get(opts, :timeout, 5_000)
      )
    else
      :error -> {:error, :client_required}
    end
  end

  @doc """
  Executes a linearizable state-machine read without appending to the VSR log.

  The current primary confirms its view with a quorum before evaluating the
  read. Singleton groups satisfy that quorum locally.
  """
  @spec read(term(), term(), keyword()) :: {:ok, term()} | {:error, term()}
  def read(group_id, operation, opts) do
    with {:ok, replica_id} <- Keyword.fetch(opts, :replica_id),
         {:ok, replicas} <- Keyword.fetch(opts, :replicas),
         {:ok, primary_id} <- primary(group_id, replica_id),
         {:ok, primary} <- find_replica(replicas, primary_id) do
      primary
      |> then(&Replica.read({group_id, &1}, operation, timeout: Keyword.get(opts, :timeout, 5_000)))
      |> maybe_retry_read(group_id, replicas, operation, opts)
    else
      :error -> {:error, :replica_identity_and_members_required}
      {:error, _reason} = error -> error
    end
  end

  defp maybe_retry_read(
         {:error, {:not_primary, primary_id}},
         group_id,
         replicas,
         operation,
         opts
       ) do
    with {:ok, primary} <- find_replica(replicas, primary_id) do
      Replica.read({group_id, primary}, operation, timeout: Keyword.get(opts, :timeout, 5_000))
    end
  end

  defp maybe_retry_read(result, _group_id, _replicas, _operation, _opts), do: result

  defp find_replica(replicas, replica_id) when is_list(replicas) do
    case Enum.find(replicas, &(&1.id == replica_id)) do
      nil -> {:error, :primary_not_configured}
      replica -> {:ok, replica}
    end
  end

  defp configuration(opts) do
    case Keyword.fetch(opts, :configuration) do
      {:ok, %Configuration{} = configuration} -> Configuration.validate(configuration)
      {:ok, attributes} -> Configuration.new(attributes)
      :error -> Configuration.new(Map.new(Keyword.take(opts, [:group_id, :replica_id, :members])))
    end
  end
end