finch

defmodule Finch.Conn do
  @moduledoc false

  alias Finch.Response
  alias Mint.HTTP
  alias Finch.Telemetry

  def new(scheme, host, port, opts, parent) do
    %{
      scheme: scheme,
      host: host,
      port: port,
      opts: opts.conn_opts,
      parent: parent,
      last_checkin: System.monotonic_time(),
      mint: nil
    }
  end

  def connect(%{mint: mint} = conn) when not is_nil(mint) do
    meta = %{
      scheme: conn.scheme,
      host: conn.host,
      port: conn.port
    }

    Telemetry.event(:reused_connection, %{}, meta)
    {:ok, conn}
  end

  def connect(conn) do
    meta = %{
      scheme: conn.scheme,
      host: conn.host,
      port: conn.port
    }

    start_time = Telemetry.start(:connect, meta)
    conn_opts = Keyword.merge(conn.opts, mode: :passive)

    case HTTP.connect(conn.scheme, conn.host, conn.port, conn_opts) do
      {:ok, mint} ->
        Telemetry.stop(:connect, start_time, meta)
        {:ok, %{conn | mint: mint}}

      {:error, error} ->
        meta = Map.put(meta, :error, error)
        Telemetry.stop(:connect, start_time, meta)
        {:error, conn, error}
    end
  end

  def open?(%{mint: nil}), do: false
  def open?(%{mint: mint}), do: HTTP.open?(mint)

  def set_mode(%{mint: nil}, _), do: {:error, "Connection is dead"}

  def set_mode(conn, mode) when mode in [:active, :passive] do
    case HTTP.set_mode(conn.mint, mode) do
      {:ok, mint} -> {:ok, %{conn | mint: mint}}
      _ -> {:error, "Connection is dead"}
    end
  end

  def stream(%{mint: nil}, _), do: {:error, "Connection is dead"}

  def stream(conn, message) do
    with {:ok, mint, responses} <- HTTP.stream(conn.mint, message) do
      {:ok, %{conn | mint: mint}, responses}
    end
  end

  def transfer(%{mint: nil}, _), do: {:error, "Connection is dead"}

  def transfer(conn, pid) do
    with {:ok, _mint} <- HTTP.controlling_process(conn.mint, pid) do
      # HTTP.controlling_process causes a side-effect, it doesn't actually change
      # the conn, so we can ignore the value returned above.
      :ok
    end
  end

  def request(%{mint: nil} = conn, _, _), do: {:error, conn, "Could not connect"}

  def request(conn, req, receive_timeout) do
    full_path = request_path(req)

    metadata = %{
      scheme: conn.scheme,
      host: conn.host,
      port: conn.port,
      path: full_path
    }

    start_time = Telemetry.start(:request, metadata)

    case HTTP.request(conn.mint, req.method, full_path, req.headers, req.body) do
      {:ok, mint, ref} ->
        Telemetry.stop(:request, start_time, metadata)
        receive_response(%{conn | mint: mint}, ref, receive_timeout, metadata)

      {:error, mint, error} ->
        metadata = Map.put(metadata, :error, error)
        Telemetry.stop(:request, start_time, metadata)
        {:error, %{conn | mint: mint}, error}
    end
  end

  def close(%{mint: nil} = conn), do: conn

  def close(conn) do
    {:ok, mint} = HTTP.close(conn.mint)
    %{conn | mint: mint}
  end

  defp request_path(%{path: path, query: nil}), do: path
  defp request_path(%{path: path, query: query}), do: "#{path}?#{query}"

  defp receive_response(conn, ref, timeout, metadata) do
    start_time = Telemetry.start(:response, metadata)

    case receive_response([], conn, ref, %Response{}, timeout) do
      {:ok, _, _} = resp ->
        Telemetry.stop(:response, start_time, metadata)
        resp

      {:error, _, error} = resp ->
        metadata = Map.put(metadata, :error, error)
        Telemetry.stop(:response, start_time, metadata)
        resp
    end
  end

  defp receive_response([], conn, ref, response, timeout) do
    case HTTP.recv(conn.mint, 0, timeout) do
      {:ok, mint, entries} ->
        receive_response(entries, %{conn | mint: mint}, ref, response, timeout)

      {:error, mint, error, _responses} ->
        {:error, %{conn | mint: mint}, error}
    end
  end

  defp receive_response([entry | entries], conn, ref, response, timeout) do
    case entry do
      {:status, ^ref, value} ->
        response = %{response | status: value}
        receive_response(entries, conn, ref, response, timeout)

      {:headers, ^ref, value} ->
        response = %{response | headers: value}
        receive_response(entries, conn, ref, response, timeout)

      {:data, ^ref, data} ->
        response = append_data_chunk(response, data)
        receive_response(entries, conn, ref, response, timeout)

      {:done, ^ref} ->
        {:ok, conn, response}

      {:error, ^ref, error} ->
        {:error, conn, error}
    end
  end

  defp append_data_chunk(%Response{body: body} = response, data_chunk) when is_binary(body) do
    %{response | body: body <> data_chunk}
  end
end