defmodule GrpcClient.Connection do
  # see the very similar original implementation of this process architecture
  # from the Mint documentation:
  # https://github.com/elixir-mint/mint/blob/796b8db097d69ede7163acba223ab2045c2773a4/pages/Architecture.md
  # This file is mostly based on https://github.com/NFIBrokerage/spear/blob/9177196721d943fda2ee1a70698580b036f220ee/lib/spear/connection.ex

  use Connection
  require Logger

  alias GrpcClient.Connection.KeepAliveTimer
  alias GrpcClient.Connection.Config
  alias GrpcClient.Connection.Request

  @post "POST"
  @closed %Mint.TransportError{reason: :closed}

  defstruct [:config, :conn, requests: %{}, keep_alive_timer: %KeepAliveTimer{}]

  @type t :: pid() | GenServer.name()

  @doc false
  def child_spec(init_arg) do
    %{
      id: __MODULE__,
      start: {__MODULE__, :start_link, [init_arg]}
    }
  end

  @doc """
  Starts a connection process

  This function can be called directly in order to link it to the current
  process, but the more common workflow is to start a `Spear.Connection`
  GenServer as a part of a supervision tree.

  ## Examples

  E.g. in an application's supervision tree defined in
  `lib/my_app/application.ex`:

      children = [
        {GrpcClient.Connection, name: MyConnection}
      ]
      Supervisor.start_link(children, strategy: :one_for_one)
  """
  @spec start_link(opts :: Keyword.t()) :: {:ok, t()} | GenServer.on_start()
  def start_link(opts) do
    name = Keyword.take(opts, [:name])
    rest = Keyword.delete(opts, :name)

    Connection.start_link(__MODULE__, rest, name)
  end

  @impl Connection
  def init(opts) do
    {:connect, :init, %__MODULE__{config: Config.new(opts)}}
  end

  @impl Connection
  def connect(_, s) do
    case do_connect(s.config) do
      {:ok, conn} ->
        {:ok, %__MODULE__{s | conn: conn, keep_alive_timer: KeepAliveTimer.start(s.config)}}

      {:error, _reason} ->
        # TODO
        {:backoff, 500, s}
    end
  end

  @impl Connection
  def disconnect(info, %__MODULE__{conn: conn} = s) do
    {:ok, _conn} = Mint.HTTP.close(conn)

    :ok = close_requests(s)

    s = %__MODULE__{
      s
      | conn: nil,
        requests: %{},
        keep_alive_timer: KeepAliveTimer.clear(s.keep_alive_timer)
    }

    case info do
      {:close, from} ->
        Connection.reply(from, {:ok, :closed})

        {:noconnect, s}

      _ ->
        {:connect, :reconnect, s}
    end
  end

  @impl Connection
  def handle_cast(:connect, s), do: {:connect, s.config, s}

  def handle_cast({:push, request_ref, message}, s) when is_reference(request_ref) do
    with {wire_data, _size} = GrpcClient.Encoding.to_binary_data(message),
         {:ok, conn} <- Mint.HTTP2.stream_request_body(s.conn, request_ref, wire_data) do
      {:noreply, put_in(s.conn, conn)}
    else
      nil ->
        {:noreply, s}

      {:error, conn, reason} ->
        s = put_in(s.conn, conn)

        if reason == @closed, do: {:disconnect, :closed, s}, else: {:noreply, s}
    end
  end

  @impl Connection
  def handle_call(_call, _from, %__MODULE__{conn: nil} = s) do
    {:reply, {:error, :closed}, s}
  end

  def handle_call(:close, from, s), do: {:disconnect, {:close, from}, s}

  def handle_call(:ping, from, s) do
    case Mint.HTTP2.ping(s.conn) do
      {:ok, conn, request_ref} ->
        s = put_in(s.conn, conn)
        s = put_in(s.requests[request_ref], {:ping, from})
        # put request ref
        {:noreply, s}

      {:error, conn, @closed} ->
        {:disconnect, :closed, {:error, :closed}, put_in(s.conn, conn)}

      {:error, conn, reason} ->
        {:reply, {:error, reason}, put_in(s.conn, conn)}
    end
  end

  def handle_call({:cancel, request_ref}, _from, s) when is_reference(request_ref) do
    with true <- Map.has_key?(s.requests, request_ref),
         {:ok, conn} <- Mint.HTTP2.cancel_request(s.conn, request_ref) do
      {:reply, :ok, put_in(s.conn, conn)}
    else
      false ->
        # idempotent success when the request_ref is not active
        {:reply, :ok, s}

      {:error, conn, @closed} ->
        {:disconnect, :closed, {:error, :closed}, put_in(s.conn, conn)}

      {:error, conn, reason} ->
        {:reply, {:error, reason}, put_in(s.conn, conn)}
    end
  end

  def handle_call({{:subscription, _, _} = type, request}, _from, s) do
    with {:ok, {s, request}} <- request_and_stream_body(s, request, nil, type) do
      {:reply, {:ok, request.request_ref}, s}
    else
      {:error, s, @closed} ->
        {:disconnect, :closed, {:error, :closed}, s}

      {:error, s, reason} ->
        {:reply, {:error, reason}, s}
    end
  end

  def handle_call({:close_subscription, request_ref}, _from, s) do
    with {:ok, conn} <- Mint.HTTP2.stream_request_body(s.conn, request_ref, :eof) do
      {:reply, :ok, put_in(s.conn, conn)}
    else
      {:error, s, @closed} ->
        {:disconnect, :closed, {:error, :closed}, s}

      {:error, s, reason} ->
        {:reply, {:error, reason}, s}
    end
  end

  def handle_call({:request, request}, from, s) do
    with {:ok, {s, _request}} <- request_and_stream_body(s, request, from, :request) do
      {:noreply, s}
    else
      {:error, s, @closed} ->
        {:disconnect, :closed, {:error, :closed}, s}

      {:error, s, reason} ->
        {:reply, {:error, reason}, s}
    end
  end

  @impl Connection
  def handle_info({:DOWN, monitor_ref, :process, _object, _reason}, s) do
    with {:ok, %{request_ref: request_ref} = request} <-
           fetch_subscription(s, monitor_ref),
         {^request, s} <- pop_in(s.requests[request_ref]),
         {:ok, conn} <- Mint.HTTP2.cancel_request(s.conn, request_ref) do
      {:noreply, put_in(s.conn, conn)}
    else
      {:error, conn, reason} ->
        s = put_in(s.conn, conn)

        if reason == @closed, do: {:disconnect, :closed, s}, else: {:noreply, s}

      _ ->
        {:noreply, s}
    end
  end

  def handle_info(:keep_alive, s) do
    case Mint.HTTP2.ping(s.conn) do
      {:ok, conn, request_ref} ->
        s = put_in(s.conn, conn)
        s = update_in(s.keep_alive_timer, &KeepAliveTimer.start_timeout_timer(&1, request_ref))

        {:noreply, s}

      {:error, conn, reason} ->
        s = put_in(s.conn, conn)

        if reason == @closed, do: {:disconnect, :closed, s}, else: {:noreply, s}
    end
  end

  def handle_info(:keep_alive_expired, s), do: {:disconnect, :keep_alive_timeout, s}

  def handle_info(message, s) do
    with %Mint.HTTP2{} = conn <- s.conn,
         {:ok, conn, responses} <- Mint.HTTP2.stream(conn, message) do
      {:noreply, put_in(s.conn, conn) |> handle_responses(responses)}
    else
      {:error, conn, reason, responses} ->
        s = put_in(s.conn, conn) |> handle_responses(responses)

        # TODO error handling
        if reason == @closed, do: {:disconnect, :closed, s}, else: {:noreply, s}

      # unknown message / no active conn in state
      _ ->
        {:noreply, s}
    end
  end

  @spec handle_responses(%__MODULE__{}, list()) :: %__MODULE__{}
  defp handle_responses(s, responses) do
    s = update_in(s.keep_alive_timer, &KeepAliveTimer.reset_interval_timer/1)

    responses
    |> Enum.reduce(s, &process_response/2)
    |> Request.continue_requests()
  end

  defp process_response({:status, request_ref, status}, s) do
    put_in(s.requests[request_ref].response.status, status)
  end

  defp process_response({:headers, request_ref, new_headers}, s) do
    update_in(
      s.requests[request_ref].response.headers,
      fn headers -> headers ++ new_headers end
    )
  end

  defp process_response({:data, request_ref, new_data}, s) do
    update_in(
      s.requests[request_ref],
      &Request.handle_data(&1, new_data)
    )
  end

  defp process_response({:pong, request_ref}, s) do
    case pop_in(s.requests[request_ref]) do
      {{:ping, from}, s} ->
        # ping was initiated by a GenServer.call/3
        Connection.reply(from, :pong)

        s

      {nil, s} ->
        # ping was initiated by the keepalive timer
        update_in(s.keep_alive_timer, &KeepAliveTimer.clear_after_timer(&1, request_ref))
    end
  end

  defp process_response({:done, request_ref}, s) do
    {request, s} = pop_in(s.requests[request_ref])

    case request do
      %{type: {:subscription, subscriber, _through}, from: nil} ->
        send(subscriber, {:eos, request_ref, :dropped})

      %{from: from, response: response} ->
        Connection.reply(from, {:ok, response})
    end

    s
  end

  defp process_response(_unknown, s), do: s

  defp request_and_stream_body(s, request, from, request_type) do
    with {:ok, conn, request_ref} <-
           Mint.HTTP2.request(s.conn, @post, request.path, request.headers, :stream),
         request = Request.new(request, request_ref, from, request_type),
         s = put_in(s.conn, conn),
         s = put_in(s.requests[request_ref], request),
         {:ok, s} <- Request.emit_messages(s, request) do
      {:ok, {s, request}}
    else
      {:error, %__MODULE__{} = s, reason} ->
        {:error, s, reason}

      {:error, conn, reason} ->
        {:error, put_in(s.conn, conn), reason}
    end
  end

  defp do_connect(config) do
    {:ok, conn} = Mint.HTTP.connect(config.scheme, config.host, config.port, config.mint_opts)

    if config.ssl_key_log_file do
      socket = Mint.HTTP.get_socket(conn)

      with file <- File.open!(config.ssl_key_log_file, [:append]),
           {:ok, [{:keylog, keylog_items}]} <- :ssl.connection_information(socket, [:keylog]) do
        for keylog_item <- keylog_items do
          :ok = IO.puts(file, keylog_item)
        end
      end
    end

    {:ok, conn}
  end

  defp close_requests(s) do
    :ok = s.requests |> Map.values() |> Enum.each(&close_request/1)
  end

  defp close_request(%{
         type: {:subscription, proc, _through},
         from: nil,
         request_ref: request_ref
       }) do
    send(proc, {:eos, request_ref, :closed})
  end

  defp close_request(%{type: _, from: from}) do
    Connection.reply(from, {:error, :closed})
  end

  @doc false
  @spec fetch_subscription(%__MODULE__{}, reference()) :: {:ok, Request.t()} | :error
  def fetch_subscription(s, monitor_ref) do
    Enum.find_value(s.requests, :error, fn {_request_ref, request} ->
      request.monitor_ref == monitor_ref && {:ok, request}
    end)
  end
end