cables

defmodule Cables do
  @moduledoc """
  Asynchronous multiplexed HTTP/2 connection manager.

  Create a new Cable using `Cable.ensure_connection/2`. Cables will not open a connection
  until the first request is recieved.

  ## Examples
      {:ok, cable} = Cables.ensure_connection("https://httpbin.org/")
      {:ok, %Cables.Response{status: 200}} = Cables.get(cable, "/get")
  """

  @type profile() :: [
    pool_timeout: integer(),
    conn_timeout: integer(),
    max_streams: integer(),
    pool_size: integer(),
    max_overflow: integer(),
    conn_ttl: integer(),
    conn_opts: map()
  ]
  @type http_method() :: :get | :post | :head | :put | :patch | :options | :delete | String.t

  defmodule Request do
    @moduledoc """
    Holds request info
    """
    defstruct [:method, :path, :headers, :body, :reply_to]
  end

  defmodule Cable do
    @moduledoc """
    Holds timeout info and poolname
    """
    defstruct [:pool_name, :pool_timeout, :conn_timeout]
  end

  @doc """
  If a pool is not already created for the specified uri, create one.
  """
  @spec ensure_connection(String.t(), atom()) :: Cabel.t()
  def ensure_connection(uri, profile_name \\ :default) do
    %URI{host: host, scheme: scheme, port: port} = URI.parse(uri)
    pool_name = String.to_atom("#{profile_name}@#{uri}")

    profile = get_profile(profile_name)
    config = poolboy_config(pool_name, profile)
    conn_opts = conn_opts(scheme, profile)

    supervisor_result = DynamicSupervisor.start_child(
      Cabels.ConnPoolSupervisor,
      :poolboy.child_spec(pool_name, config, [to_charlist(host), port, conn_opts, Keyword.get(profile, :max_streams), Keyword.get(profile, :conn_ttl)])
    )
    case supervisor_result do
      {:ok, _pid} ->
        {:ok, make_cable(pool_name, profile)}
      {:error, {:already_started, _pid}} ->
        {:ok, make_cable(pool_name, profile)}
    end
  end

  @doc """
  Start a request and handle it with the request_handler. `Cables.Handler`
  This gives you full control over sending and recieving stream data.

  For an example see `Cables.Response`.

  ## Examples

      iex> {:ok, cable} = Cables.ensure_connection("https://httpbin.org/")
      ...> {:ok, %Cables.Response{status: status}} = Cables.request(cable, :get, "/get", Cables.Response, nil)
      ...> status
      200
  """
  @spec request(Cabel.t(), http_method(), String.t, [{String.t, String.t}], String.t, pid(), module(), any()) :: t::any()
  def request(%Cable{pool_name: pool_name, pool_timeout: pool_timeout, conn_timeout: timeout}, method, path, headers \\ [], body \\ "", reply_to \\ nil, module, init_args) do
    method_str =
      case method do
        :get -> "GET"
        :post -> "POST"
        :head -> "HEAD"
        :put -> "PUT"
        :patch -> "PATCH"
        :delete -> "DELETE"
        :options -> "OPTIONS"
        m when is_binary(m) -> m
      end

    req = %Request{
      method: method_str,
      path: path,
      headers: headers,
      body: body,
      reply_to: if is_nil(reply_to) do self() else reply_to end
    }

    c = :poolboy.checkout(pool_name, true, pool_timeout)
    {{remote_conn, stream_ref}, check_in} =
      try do
        GenServer.call(c, {:request, req})
      rescue
        exception ->
          :ok = :poolboy.checkin(pool_name, c)
          reraise exception, __STACKTRACE__
      else
        {ret, streams_available} ->
          # If there are available streams, immedidately return to the pool. Else return
          # to the pool after this stream is finished.
          if streams_available do
            :ok = :poolboy.checkin(pool_name, c)
          end
          {ret, not streams_available}
      end

    try do
      module.handle(remote_conn, stream_ref, timeout, init_args)
    after
      if check_in do
        :ok = :poolboy.checkin(pool_name, c)
      end
      GenServer.cast(c, {:finish_stream, stream_ref})
    end
  end

  @doc """
  Simple GET request with `Cables.Response`

  ## Examples

      iex> {:ok, cable} = Cables.ensure_connection("https://httpbin.org/")
      ...> {:ok, %Cables.Response{status: status}} = Cables.get(cable, "/get")
      ...> status
      200

  """
  @spec get(Cabel.t(), [{String.t, String.t}], String.t, pid()) :: {:ok, Cables.Response.t} | {:error, any()}
  def get(cable, path, headers \\ [], reply_to \\ nil) do
    request(cable, :get, path, headers, "", reply_to, Cables.Response, nil)
  end

  @doc """
  Simple POST request with `Cables.Response`

  ## Examples

      iex> {:ok, cable} = Cables.ensure_connection("https://httpbin.org/")
      ...> {:ok, %Cables.Response{status: status}} = Cables.post(cable, "/post", [], "hello world")
      ...> status
      200

  """
  @spec post(Cabel.t(), String.t, [{String.t, String.t}], iodata(), pid()) :: {:ok, Cables.Response.t} | {:error, any()}
  def post(cable, path, headers \\ [], body \\ "", reply_to \\ nil) do
    request(cable, :post, path, headers, body, reply_to, Cables.Response, nil)
  end

  @doc """
  Simple PUT request with `Cables.Response`

  ## Examples

      iex> {:ok, cable} = Cables.ensure_connection("https://httpbin.org/")
      ...> {:ok, %Cables.Response{status: status}} = Cables.put(cable, "/put", [], "hello world")
      ...> status
      200
  """
  @spec put(Cabel.t(), String.t, [{String.t, String.t}], iodata(), pid()) :: {:ok, Cables.Response.t} | {:error, any()}
  def put(cable, path, headers \\ [], body \\ "", reply_to \\ nil) do
    request(cable, :put, path, headers, body, reply_to, Cables.Response, nil)
  end

  @doc """
  Simple PATCH request with `Cables.Response`

  ## Examples

      iex> {:ok, cable} = Cables.ensure_connection("https://httpbin.org/")
      ...> {:ok, %Cables.Response{status: status}} = Cables.patch(cable, "/patch", [], "hello world")
      ...> status
      200
  """
  @spec patch(Cabel.t(), String.t, [{String.t, String.t}], iodata(), pid()) :: {:ok, Cables.Response.t} | {:error, any()}
  def patch(cable, path, headers \\ [], body \\ "", reply_to \\ nil) do
    request(cable, :patch, path, headers, body, reply_to, Cables.Response, nil)
  end

  @doc """
  Simple DELETE request with `Cables.Response`

  ## Examples

      iex> {:ok, cable} = Cables.ensure_connection("https://httpbin.org/")
      ...> {:ok, %Cables.Response{status: status}} = Cables.delete(cable, "/delete")
      ...> status
      200
  """
  @spec delete(Cabel.t(), String.t, [{String.t, String.t}], iodata(), pid()) :: {:ok, Cables.Response.t} | {:error, any()}
  def delete(cable, path, headers \\ [], body \\ "", reply_to \\ nil) do
    request(cable, :delete, path, headers, body, reply_to, Cables.Response, nil)
  end

  @doc """
  Simple HEAD request with `Cables.Response`
  """
  @spec head(Cabel.t(), [{String.t, String.t}], String.t, pid()) :: {:ok, Cables.Response.t} | {:error, any()}
  def head(cable, path, headers \\ [], reply_to \\ nil) do
    request(cable, :head, path, headers, "", reply_to, Cables.Response, nil)
  end

  @doc """
  Simple OPTIONS request with `Cables.Response`
  """
  @spec options(Cabel.t(), [{String.t, String.t}], String.t, pid()) :: {:ok, Cables.Response.t} | {:error, any()}
  def options(cable, path, headers \\ [], reply_to \\ nil) do
    request(cable, :options, path, headers, reply_to, Cables.Response, nil)
  end

  @doc """
  Send a piece of data. Make sure to use &send_final_data/3 to send the final chunk.
  """
  @spec send_data(pid(), reference(), String.t) :: :ok
  def send_data(gun_pid, stream_ref, data) do
    :gun.data(gun_pid, stream_ref, :nofin, data)
  end

  @doc """
  Send a piece of data and indicate that the request body has finished.
  """
  @spec send_final_data(pid(), reference(), String.t) :: :ok
  def send_final_data(gun_pid, stream_ref, data) do
    :gun.data(gun_pid, stream_ref, :fin, data)
  end

  @spec conn_opts(String.t(), profile()) :: any()
  defp poolboy_config(pool_name, profile) do
    [
      name: {:local, pool_name},
      worker_module: Cables.Connection,
      size: Keyword.get(profile, :pool_size),
      max_overflow: Keyword.get(profile, :max_overflow),
      strategy: :lifo
    ]
  end

  @spec conn_opts(String.t(), profile()) :: map()
  defp conn_opts(scheme, profile) do
    transport =
      case scheme do
        "https" -> :tls
        _ -> :tcp
      end
    Map.merge(%{transport: transport}, Keyword.get(profile, :conn_opts))
  end

  @spec make_cable(atom(), profile()) :: Cable.t()
  defp make_cable(pool_name, profile) do
    %Cable{
      pool_name: pool_name,
      pool_timeout: Keyword.get(profile, :pool_timeout),
      conn_timeout: Keyword.get(profile, :conn_timeout),
    }
  end

  @spec get_profile(atom()) :: profile()
  defp get_profile(profile_name) do
    default = [
      pool_timeout: 5_000,
      conn_timeout: 5_000,
      max_streams: 100,
      pool_size: 10,
      max_overflow: 0,
      conn_ttl: 10_000,
      conn_opts: %{}
    ]
    profiles = Application.get_env(:cables, :profiles, [])
    Keyword.merge(default, Keyword.get(profiles, profile_name, []))
  end
end