defmodule K8s.Client.Mint.HTTPAdapter do
  @moduledoc """
  The Mint client implementation. This module handles both, HTTP requests and
  websocket connections.

  ## Processes

  The module creates a process per connection to the Kubernetes API server.
  It supports `HTTP/2` for HTTP requests, but not for websockets. So while
  an open connection can process multiple requests (if the server supports
  `HTTP/2`), it can only process one single websocket connection.
  Therefore, each websocket connection is handled in its own process.

  ## State

  The module keeps track of the `Mint.HTTP` connection struct and a map of
  pending requests (`K8s.Client.Mint.Request`) for that connection, indexed by the
  `Mint.Types.request_ref()`.

  ### Requests

  Requests are calls to the GenServer that immediately return while the GenServer
  receives the response parts in the background. The way these response parts are
  returned depends on the `state_to` argument passed to `request/7` resp.
  `websocket_request/5`. See these function's docs for more details.
  """
  use GenServer, restart: :temporary

  alias K8s.Client.HTTPError
  alias K8s.Client.Mint.Request

  import K8s.Sys.Logger, only: [log_prefix: 1]
  require Logger
  require Mint.HTTP

  # healthcheck frequency in seconds
  @healthcheck_freq 30

  @type connection_args_t ::
          {scheme :: atom(), host :: binary(), port :: integer(), opts :: keyword()}

  @typedoc """
  Describes the state of the connection.

  - `:conn` - The current state of the `Mint` connection.
  - `:requests` - The opened requests over this connection (Only `HTTP/2` connections will hold multiple entries in this field.)
  """
  @type t :: %__MODULE__{
          conn: Mint.HTTP.t(),
          requests: %{reference() => Request.t()}
        }

  defstruct [:conn, requests: %{}]

  @doc """
  Opens a connection to Kubernetes, defined by `uri` and `opts`,
  and starts the GenServer.
  """
  @spec start_link(connection_args_t()) :: GenServer.on_start()
  def start_link(conn_args) do
    GenServer.start_link(__MODULE__, conn_args)
  end

  @doc """
  Returns connection arguments for the given `URI` and HTTP options.
  """
  @spec connection_args(URI.t(), keyword()) :: connection_args_t()
  def connection_args(uri, opts) do
    {String.to_atom(uri.scheme), uri.host, uri.port, opts}
  end

  @spec open?(GenServer.server(), :read | :write | :read_write) :: boolean()
  def open?(pid, type \\ :read_write) do
    GenServer.call(pid, {:open?, type})
  catch
    :exit, _ -> false
  end

  @doc """
  Starts a HTTP request. The way the response parts are returned depends on the
  `stream_to` argument passed to it.

    - `nil` - response parts are buffered. In order to receive them, the caller
      needs to `recv/2` passing it the `request_ref` returned by this function.
    - `pid` - response parts are sent to the process with the given `pid`.
    - `{pid, reference}` - response parts are sent to the process with the given
      `pid`. Messages are of the form `{reference, response_part}`.
  """
  @spec request(
          GenServer.server(),
          method :: binary(),
          path :: binary(),
          Mint.Types.headers(),
          body :: iodata() | nil | :stream,
          pool :: pid() | nil,
          stream_to :: pid() | nil
        ) :: {:ok, reference()} | {:error, HTTPError.t()}
  def request(pid, method, path, headers, body, pool, stream_to) do
    GenServer.call(pid, {:request, method, path, headers, body, pool, stream_to})
  end

  @doc """
  Upgrades to a Websocket connection. The way the frames are returned depends
  on the `stream_to` argument passed to it.

    - `nil` - frames are buffered. In order to receive them, the caller
      needs to `recv/2` passing it the `request_ref` returned by this function.
    - `pid` - frames are sent to the process with the given `pid`.
    - `{pid, reference}` - frames are sent to the process with the given
      `pid`. Messages are of the form `{reference, response_part}`.
  """
  @spec websocket_request(
          pid(),
          path :: binary(),
          Mint.Types.headers(),
          pool :: pid() | nil,
          stream_to :: pid() | nil
        ) :: {:ok, reference()} | {:error, HTTPError.t()}
  def websocket_request(pid, path, headers, pool, stream_to) do
    GenServer.call(pid, {:websocket_request, path, headers, pool, stream_to})
  end

  @doc """
  Returns response parts / frames that were buffered by the process. The
  `stream_to` must have been set to `nil` in `request/7` or
  `websocket_request/5`.

  If the buffer is empty, this call blocks until the next response part /
  frame is received.
  """
  @spec recv(GenServer.server(), reference()) :: list()
  def recv(pid, ref) do
    GenServer.call(pid, {:recv, ref}, :infinity)
  end

  @doc """
  Sends the given `data` throught the websocket specified by the `request_ref`.
  """
  @spec websocket_send(
          GenServer.server(),
          reference(),
          term()
        ) :: :ok
  def websocket_send(pid, request_ref, data) do
    GenServer.cast(pid, {:websocket_send, request_ref, data})
  end

  @impl true
  def init({scheme, host, port, opts}) do
    case Mint.HTTP.connect(scheme, host, port, opts) do
      {:ok, conn} ->
        Process.send_after(self(), :healthcheck, @healthcheck_freq * 1_000)
        state = %__MODULE__{conn: conn}
        {:ok, state}

      {:error, error} ->
        Logger.error(log_prefix("Failed initializing HTTPAdapter GenServer"), library: :k8s)
        {:stop, HTTPError.from_exception(error)}
    end
  end

  @impl true
  def handle_call({:open?, type}, _from, state) do
    {:reply, Mint.HTTP.open?(state.conn, type), state}
  end

  def handle_call({:request, method, path, headers, body, pool, stream_to}, from, state) do
    caller_ref = from |> elem(0) |> Process.monitor()
    conn = state.conn

    # For HTTP2, if the body is larger than the connection window, we've got to
    # stream it to the server.
    {body, pending_request_body} =
      cond do
        Mint.HTTP.protocol(conn) == :http1 -> {body, nil}
        is_nil(body) -> {nil, nil}
        :otherwise -> {:stream, body}
      end

    with {:ok, conn, request_ref} <- Mint.HTTP.request(conn, method, path, headers, body),
         {:request, request} <-
           {:request,
            Request.new(
              request_ref: request_ref,
              pool: pool,
              stream_to: stream_to,
              caller_ref: caller_ref,
              pending_request_body: pending_request_body
            )},
         {:ok, request, conn} <- Request.stream_request_body(request, conn) do
      state = put_in(state.requests[request_ref], request) |> struct!(conn: conn)
      {:reply, {:ok, request_ref}, state}
    else
      {:error, conn, error} ->
        state = struct!(state, conn: conn)

        {:reply, {:error, HTTPError.from_exception(error)}, state}
    end
  end

  def handle_call({:websocket_request, path, headers, pool, stream_to}, from, state) do
    caller_ref = from |> elem(0) |> Process.monitor()

    with {:ok, conn} <- Mint.HTTP.set_mode(state.conn, :passive),
         {:ok, conn, request_ref} <- Mint.WebSocket.upgrade(:wss, conn, path, headers),
         {:ok, conn, response} <- Request.receive_upgrade_response(conn, request_ref),
         {:ok, conn} <- Mint.HTTP.set_mode(conn, :active),
         {:ok, conn, websocket} <-
           Mint.WebSocket.new(conn, request_ref, response.status, response.headers) do
      state =
        put_in(
          state.requests[request_ref],
          Request.new(
            request_ref: request_ref,
            websocket: websocket,
            pool: pool,
            stream_to: stream_to,
            caller_ref: caller_ref
          )
        )

      send(stream_to, {:open, true})
      {:reply, {:ok, request_ref}, struct!(state, conn: conn)}
    else
      {:error, error} ->
        GenServer.reply(from, {:error, HTTPError.from_exception(error)})
        {:stop, :normal, state}

      {:error, conn, error} ->
        GenServer.reply(from, {:error, HTTPError.from_exception(error)})
        {:stop, :normal, struct!(state, conn: conn)}
    end
  end

  def handle_call({:recv, request_ref}, from, state) do
    {_, state} =
      get_and_update_in(
        state.requests[request_ref],
        &Request.recv(&1, from)
      )

    shutdown_if_closed(state)
  end

  @impl true
  def handle_cast({:websocket_send, request_ref, data}, state) do
    request = state.requests[request_ref]

    with {:ok, frame} <- Request.map_outgoing_frame(data),
         {:ok, websocket, data} <- Mint.WebSocket.encode(request.websocket, frame),
         {:ok, conn} <- Mint.WebSocket.stream_request_body(state.conn, request_ref, data) do
      state = struct!(state, conn: conn)
      state = put_in(state.requests[request_ref].websocket, websocket)
      {:noreply, state}
    end
  end

  @impl true
  def handle_info(message, %__MODULE__{conn: conn} = state)
      when Mint.HTTP.is_connection_message(conn, message) do
    with {:ok, conn, responses} <- Mint.WebSocket.stream(state.conn, message),
         {:ok, state} <- stream_pending_request_bodies(struct!(state, conn: conn)) do
      process_responses_or_frames(state, responses)
    else
      {:error, conn, %Mint.TransportError{reason: :closed}, []} ->
        Logger.debug("The connection was closed.", library: :k8s)

        shutdown_if_closed(struct!(state, conn: conn))

      {:error, conn, error} ->
        Logger.warning(
          log_prefix(
            "An error occurred when streaming the request body: #{Exception.message(error)}"
          ),
          error: error,
          library: :k8s
        )

        struct!(state, conn: conn)

      {:error, conn, error, responses} ->
        Logger.warning(
          log_prefix(
            "An error occurred when streaming the response: #{Exception.message(error)}"
          ),
          error: error,
          library: :k8s
        )

        state
        |> struct!(conn: conn)
        |> process_responses_or_frames(responses)
    end
  end

  def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
    state =
      state.requests
      |> Map.filter(fn {_request_ref, request} -> request.caller_ref == ref end)
      |> Map.keys()
      |> Enum.reduce_while(state, fn
        request_ref, state ->
          case pop_in(state.requests[request_ref]) do
            {%Request{}, %{conn: %Mint.HTTP2{}} = state} ->
              conn = Mint.HTTP2.cancel_request(state.conn, request_ref) |> elem(1)
              {:cont, struct!(state, conn: conn)}

            {_, state} ->
              {:halt, {:stop, state}}
          end
      end)

    case state do
      {:stop, state} ->
        Logger.debug(
          log_prefix(
            "Received :DOWN signal from parent process. Terminating HTTPAdapter #{inspect(self())}."
          ),
          library: :k8s
        )

        {:stop, :normal, state}

      state ->
        {:noreply, state}
    end
  end

  # This is called regularly to check whether the connection is still open. If
  # it's not open, and all buffers are emptied, this process is considered
  # garbage and is stopped.
  def handle_info(:healthcheck, state) do
    if Mint.HTTP.open?(state.conn, :read) or any_non_empty_buffers?(state) do
      Process.send_after(self(), :healthcheck, @healthcheck_freq * 1000)
      {:noreply, state}
    else
      Logger.warning(
        log_prefix("Connection closed for reading and writing - stopping this process."),
        library: :k8s
      )

      {:stop, {:shutdown, :closed}, state}
    end
  end

  @impl true
  def terminate(reason, state) do
    state = state

    state.requests
    |> Enum.each(fn
      {_request_ref, request} when is_nil(request.websocket) ->
        Request.put_response(
          request,
          {:error, reason}
        )

      {request_ref, request} ->
        {:ok, _websocket, data} = Mint.WebSocket.encode(request.websocket, :close)
        Mint.WebSocket.stream_request_body(state.conn, request_ref, data)
    end)

    Mint.HTTP.close(state.conn)

    Logger.debug(log_prefix("Terminating HTTPAdapter GenServer #{inspect(self())}"),
      library: :k8s
    )

    :ok
  end

  @spec stream_pending_request_bodies(t()) ::
          {:ok, t()} | {:error, Mint.HTTP.t(), Mint.Types.error()}
  defp stream_pending_request_bodies(state) do
    stream_pending_request_bodies(state, Map.values(state.requests))
  end

  @spec stream_pending_request_bodies(t(), [Request.t()]) ::
          {:ok, t()} | {:error, Mint.HTTP.t(), Mint.Types.error()}
  defp stream_pending_request_bodies(state, []) do
    {:ok, state}
  end

  defp stream_pending_request_bodies(state, [request | rest]) do
    case Request.stream_request_body(
           request,
           state.conn
         ) do
      {:ok, request, conn} ->
        state =
          put_in(state.requests[request.request_ref], request)
          |> struct!(conn: conn)

        stream_pending_request_bodies(state, rest)

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

  @spec process_responses_or_frames(t(), [Mint.Types.response()]) :: {:noreply, t()}
  defp process_responses_or_frames(state, [{:data, request_ref, data}] = responses) do
    request = state.requests[request_ref]

    if is_nil(request.websocket) do
      process_responses(state, responses)
    else
      {:ok, websocket, frames} = Mint.WebSocket.decode(request.websocket, data)
      state = put_in(state.requests[request_ref].websocket, websocket)
      process_frames(state, request_ref, frames)
    end
  end

  defp process_responses_or_frames(state, responses) do
    process_responses(state, responses)
  end

  @spec process_frames(t(), reference(), list(Mint.WebSocket.frame())) :: {:noreply, t()}
  defp process_frames(state, request_ref, frames) do
    state =
      frames
      |> Enum.map(&Request.map_frame/1)
      |> Enum.reduce(state, fn mapped_frame, state ->
        {_, state} =
          get_and_update_in(
            state.requests[request_ref],
            &Request.put_response(&1, mapped_frame)
          )

        state
      end)

    {:noreply, state}
  end

  @spec process_responses(t(), [Mint.Types.response()]) :: {:noreply, t()}
  defp process_responses(state, responses) do
    state =
      responses
      |> Enum.map(&Request.map_response/1)
      |> Enum.reduce(state, fn {mapped_response, request_ref}, state ->
        {_, state} =
          get_and_update_in(
            state.requests[request_ref],
            &Request.put_response(&1, mapped_response)
          )

        state
      end)

    {:noreply, state}
  end

  @spec shutdown_if_closed(t()) :: {:noreply, t()} | {:stop, {:shutdown, :closed}, t()}
  defp shutdown_if_closed(state) do
    if Mint.HTTP.open?(state.conn, :read) or any_non_empty_buffers?(state) do
      {:noreply, state}
    else
      Logger.warning(
        log_prefix("Connection closed for reading and writing - stopping this process."),
        library: :k8s
      )

      {:stop, {:shutdown, :closed}, state}
    end
  end

  @spec any_non_empty_buffers?(t()) :: boolean()
  defp any_non_empty_buffers?(state) do
    Enum.any?(state.requests, fn {_, request} -> request.buffer != [] end)
  end
end
