electric_client

defmodule Electric.Client.Fetch.Request do
  use GenServer

  alias Electric.Client.Fetch
  alias Electric.Client.Offset
  alias Electric.Client.ShapeDefinition

  require Logger

  defstruct [
    :base_url,
    :shape_id,
    :live,
    :shape,
    :next_cursor,
    update_mode: :modified,
    method: :get,
    offset: Offset.before_all()
  ]

  @type t :: %__MODULE__{
          method: :get | :head,
          base_url: String.t(),
          offset: Electric.Client.Offset.t(),
          shape_id: Electric.Client.shape_id() | nil,
          update_mode: Electric.Client.update_mode(),
          live: boolean(),
          next_cursor: Electric.Client.cursor(),
          shape: ShapeDefinition.t()
        }

  @doc false
  def name(request, fetcher) do
    request
    |> request_id(fetcher)
    |> name()
  end

  @doc false
  def name(request_id) do
    {:via, Registry, {Electric.Client.Registry, {__MODULE__, request_id}}}
  end

  defp request_id(%{base_url: base_url, shape_id: nil, shape: shape_definition}, {fetch_impl, _}) do
    {fetch_impl, base_url, shape_definition}
  end

  defp request_id(request, {fetch_impl, _}) do
    %{base_url: base_url, offset: offset, live: live, shape_id: shape_id} = request
    {fetch_impl, base_url, shape_id, Offset.to_tuple(offset), live}
  end

  @doc false
  def request(%__MODULE__{} = request, {_, _} = fetcher) do
    request_id = request_id(request, fetcher)

    # register this pid before making the request to avoid race conditions for
    # very fast responses
    {:ok, monitor_pid} = start_monitor(request_id)

    try do
      ref = Fetch.Monitor.register(monitor_pid, self())

      {:ok, _request_pid} = start_request(request_id, request, fetcher, monitor_pid)

      Fetch.Monitor.wait(ref)
    catch
      :exit, {reason, _} ->
        Logger.debug(fn ->
          "Request process ended with reason #{inspect(reason)} before we could register. Re-attempting."
        end)

        request(request, fetcher)
    end
  end

  defp start_request(request_id, request, fetcher, monitor_pid) do
    DynamicSupervisor.start_child(
      Electric.Client.RequestSupervisor,
      {__MODULE__, {request_id, request, fetcher, monitor_pid}}
    )
    |> return_existing()
  end

  defp start_monitor(request_id) do
    DynamicSupervisor.start_child(
      Electric.Client.RequestSupervisor,
      {Electric.Client.Fetch.Monitor, request_id}
    )
    |> return_existing()
  end

  defp return_existing({:ok, pid}), do: {:ok, pid}
  defp return_existing({:error, {:already_started, pid}}), do: {:ok, pid}
  defp return_existing(error), do: error

  @doc false
  def child_spec({request_id, _request, _fetcher, _monitor_pid} = args) do
    %{
      id: {__MODULE__, request_id},
      start: {__MODULE__, :start_link, [args]},
      restart: :transient,
      type: :worker
    }
  end

  @doc false
  def start_link({request_id, request, fetcher, monitor_pid}) do
    GenServer.start_link(__MODULE__, {request_id, request, fetcher, monitor_pid},
      name: name(request_id)
    )
  end

  @impl true
  def init({request_id, request, fetcher, monitor_pid}) do
    Logger.debug(fn ->
      "Starting request for #{inspect(request_id)}"
    end)

    state = %{
      request_id: request_id,
      request: request,
      fetcher: fetcher,
      monitor_pid: monitor_pid
    }

    {:ok, state, {:continue, :request}}
  end

  @impl true
  def handle_continue(:request, state) do
    %{fetcher: {module, opts}, request: request} = state

    case module.fetch(request, opts) do
      {:ok, %Fetch.Response{status: status} = response} when status in 200..299 ->
        reply(response, state)

      {:ok, %Fetch.Response{} = response} ->
        # Turn HTTP errors into errors
        reply({:error, response}, state)

      error ->
        reply(error, state)
    end

    {:stop, :normal, state}
  end

  defp reply(response, %{monitor_pid: monitor_pid}) do
    Fetch.Monitor.reply(monitor_pid, response)
  end
end