electric_client

defmodule Electric.Client.Fetch.Request do
  use GenServer

  alias Electric.Client
  alias Electric.Client.Fetch
  alias Electric.Client.Util

  require Logger

  defstruct [
    :stream_id,
    :endpoint,
    :shape_handle,
    :live,
    :next_cursor,
    # `replica` does not belong here, it's part of the shape definition that we
    # receive via the client params. the other keys in this struct are part
    # of consuming a stream rather than defining the shape.
    # kept for backwards compatibility
    replica: :default,
    method: :get,
    offset: Client.Offset.before_all(),
    params: %{},
    headers: %{},
    authenticated: false
  ]

  @type params :: %{String.t() => String.t()}
  @type headers :: %{String.t() => [String.t()] | String.t()}

  fields = [
    stream_id: quote(do: term()),
    method: quote(do: :get | :head | :delete | :options),
    endpoint: quote(do: URI.t()),
    offset: quote(do: Electric.Client.offset()),
    shape_handle: quote(do: Electric.Client.shape_handle() | nil),
    replica: quote(do: Electric.Client.replica()),
    live: quote(do: boolean()),
    next_cursor: quote(do: Electric.Client.cursor()),
    params: quote(do: params()),
    headers: quote(do: headers())
  ]

  @type unauthenticated :: %__MODULE__{unquote_splicing(fields), authenticated: false}
  @type authenticated :: %__MODULE__{unquote_splicing(fields), authenticated: true}
  @type t :: unauthenticated() | authenticated()

  # the base url should come from the client
  attrs = Keyword.delete(fields, :endpoint)

  attr_types =
    attrs
    |> Enum.reduce(nil, fn
      {name, spec}, nil -> quote(do: unquote({name, spec}))
      {name, spec}, acc -> quote(do: unquote({name, spec}) | unquote(acc))
    end)

  @type attr :: unquote(attr_types)
  @type attrs :: [attr()] | %{unquote_splicing(attrs)}

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

  @doc """
  Returns the URL for the Request.
  """
  @spec url(t()) :: binary()
  def url(%__MODULE__{} = request, opts \\ []) do
    request
    |> uri(opts)
    |> URI.to_string()
  end

  @doc """
  Returns the %URI{} for the Request.
  """
  @spec uri(t()) :: URI.t()
  def uri(%__MODULE__{} = request, opts \\ []) do
    %{endpoint: endpoint} = request

    if Keyword.get(opts, :query, true) do
      # Convert map to _ordered_ list of query parameters
      # to ensure consistent caching
      query =
        request
        |> params()
        |> Map.to_list()
        |> List.keysort(0)
        |> URI.encode_query(:rfc3986)

      %{endpoint | query: query}
    else
      endpoint
    end
  end

  @doc false
  @spec params(t()) :: params()
  def params(%__MODULE__{} = request) do
    %{
      replica: replica,
      live: live?,
      shape_handle: shape_handle,
      offset: offset,
      next_cursor: cursor,
      params: params
    } = request

    (params || %{})
    |> Map.merge(%{"offset" => to_string(offset)})
    |> Util.map_put_if("replica", to_string(replica), replica != :default)
    |> Util.map_put_if("handle", shape_handle, is_binary(shape_handle))
    |> Util.map_put_if("live", "true", live?)
    |> Util.map_put_if("cursor", to_string(cursor), !is_nil(cursor))
  end

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

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

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

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

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

  @impl true
  def handle_continue(:request, state) do
    %{client: client, request: request} = state
    %{fetch: {fetcher, fetcher_opts}} = client

    authenticated_request = Client.authenticate_request(client, request)

    try do
      case fetcher.fetch(authenticated_request, fetcher_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
    rescue
      error ->
        reply({:error, error}, state)
    end

    {:stop, :normal, state}
  end

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