defmodule LambdaEthereumConsensus.P2P.BlockDownloader do
  @moduledoc """
  This module requests blocks from peers.
  """
  require Logger

  alias LambdaEthereumConsensus.Libp2pPort
  alias LambdaEthereumConsensus.Metrics
  alias LambdaEthereumConsensus.P2P
  alias LambdaEthereumConsensus.P2P.ReqResp
  alias Types.SignedBeaconBlock
  alias Types.Store

  @blocks_by_range_protocol_id "/eth2/beacon_chain/req/beacon_blocks_by_range/2/ssz_snappy"
  @blocks_by_root_protocol_id "/eth2/beacon_chain/req/beacon_blocks_by_root/2/ssz_snappy"

  # Requests to peers might fail for various reasons,
  # for example they might not support the protocol or might not reply
  # so we want to try again with a different peer
  @default_retries 5

  @type range :: {Types.slot(), Types.slot()}
  @type download_result :: {:ok, [SignedBeaconBlock.t()]} | {:error, any()}
  @type on_blocks ::
          (Store.t(), {:ok, range(), [SignedBeaconBlock.t()]} | {:error, range(), any()} ->
             term())

  @doc """
  Requests a series of blocks in batch, and synchronously (the caller will block waiting for the
  result). As this is a synchronous function, the caller process must be different than Libp2pPort.

  Arguments:
  - slot: the slot that marks the start of the requested range.
  - count: the amount of blocks that will be requested for download.
  - retries (optional): if the download fails the request will retry, using a different random
    peer. This argument determines the amount of times that will happen before returning an error.
  """
  @spec request_blocks_by_range_sync(Types.slot(), non_neg_integer(), non_neg_integer()) ::
          download_result()
  def request_blocks_by_range_sync(slot, count, retries \\ @default_retries)

  def request_blocks_by_range_sync(_slot, 0, _retries), do: {:ok, []}

  def request_blocks_by_range_sync(slot, count, retries) do
    pid = self()

    request_blocks_by_range(
      slot,
      count,
      fn store, result -> tap(store, send(pid, result)) end,
      retries
    )

    receive do
      result -> result
    end
  end

  @spec request_blocks_by_range(Types.slot(), non_neg_integer(), on_blocks(), non_neg_integer()) ::
          :ok
  @spec request_blocks_by_range(Types.slot(), non_neg_integer(), on_blocks()) :: :ok
  def request_blocks_by_range(slot, count, on_blocks, retries \\ @default_retries)

  def request_blocks_by_range(_slot, 0, _on_blocks, _retries), do: :ok

  def request_blocks_by_range(slot, count, on_blocks, retries) do
    Logger.debug("Requesting block", slot: slot)

    peer_id = get_some_peer()

    request =
      %Types.BeaconBlocksByRangeRequest{start_slot: slot, count: count}
      |> ReqResp.encode_request()

    Libp2pPort.send_async_request(peer_id, @blocks_by_range_protocol_id, request, fn store,
                                                                                     response ->
      Metrics.handler_span(
        "response_handler",
        "blocks_by_range",
        fn ->
          handle_blocks_by_range_response(
            store,
            response,
            slot,
            count,
            retries,
            peer_id,
            on_blocks
          )
        end
      )
    end)
  end

  defp handle_blocks_by_range_response(store, response, slot, count, retries, peer_id, on_blocks) do
    with {:ok, response_message} <- response,
         {:ok, blocks} <- ReqResp.decode_response(response_message, SignedBeaconBlock),
         :ok <- verify_batch(blocks, slot, count) do
      tags = %{result: "success", type: "by_slot", reason: "success"}
      :telemetry.execute([:network, :request], %{blocks: count}, tags)
      on_blocks.(store, {:ok, {slot, slot + count - 1}, blocks})
    else
      {:error, reason} ->
        tags = %{type: "by_slot", reason: parse_reason(reason)}
        P2P.Peerbook.penalize_peer(peer_id)

        if retries > 0 do
          :telemetry.execute([:network, :request], %{blocks: 0}, Map.put(tags, :result, "retry"))
          Logger.debug("Retrying request for #{count} blocks", slot: slot)
          request_blocks_by_range(slot, count, on_blocks, retries - 1)
          {:ok, store}
        else
          :telemetry.execute([:network, :request], %{blocks: 0}, Map.put(tags, :result, "error"))
          # TODO: Add block range that failed in the reason
          on_blocks.(store, {:error, {slot, slot + count - 1}, reason})
        end
    end
  end

  @spec request_block_by_root(
          Types.root(),
          (Store.t(), {:ok, SignedBeaconBlock.t()} | {:error, binary()} -> :ok),
          integer()
        ) :: :ok
  def request_block_by_root(root, on_block, retries \\ @default_retries) do
    request_blocks_by_root(
      [root],
      fn store, response -> on_block.(store, flatten_response(response)) end,
      retries
    )
  end

  @spec request_blocks_by_root(
          [Types.root()],
          (Store.t(), {:ok, [SignedBeaconBlock.t()]} | {:error, binary()} -> :ok),
          integer()
        ) :: :ok
  def request_blocks_by_root(roots, on_blocks, retries \\ @default_retries)

  def request_blocks_by_root([], _on_blocks, _retries), do: {:ok, []}

  def request_blocks_by_root(roots, on_blocks, retries) do
    Logger.debug("Requesting block for roots #{Enum.map_join(roots, ", ", &Base.encode16/1)}")

    peer_id = get_some_peer()

    request = ReqResp.encode_request({roots, TypeAliases.beacon_blocks_by_root_request()})

    Libp2pPort.send_async_request(peer_id, @blocks_by_root_protocol_id, request, fn store,
                                                                                    response ->
      Metrics.handler_span(
        "response_handler",
        "blocks_by_root",
        fn ->
          handle_blocks_by_root_response(store, response, roots, on_blocks, peer_id, retries)
        end
      )
    end)
  end

  defp handle_blocks_by_root_response(store, response, roots, on_blocks, peer_id, retries) do
    with {:ok, response_message} <- response,
         {:ok, blocks} <- ReqResp.decode_response(response_message, SignedBeaconBlock) do
      tags = %{result: "success", type: "by_root", reason: "success"}
      :telemetry.execute([:network, :request], %{blocks: length(roots)}, tags)
      on_blocks.(store, {:ok, blocks})
    else
      {:error, reason} ->
        tags = %{type: "by_root", reason: parse_reason(reason)}
        P2P.Peerbook.penalize_peer(peer_id)

        if retries > 0 do
          :telemetry.execute([:network, :request], %{blocks: 0}, Map.put(tags, :result, "retry"))
          pretty_roots = Enum.map_join(roots, ", ", &Base.encode16/1)

          Logger.debug("Retrying request for block roots #{pretty_roots}: #{inspect(reason)}")
          request_blocks_by_root(roots, on_blocks, retries - 1)
          {:ok, store}
        else
          :telemetry.execute([:network, :request], %{blocks: 0}, Map.put(tags, :result, "error"))
          on_blocks.(store, {:error, reason})
        end
    end
  end

  defp get_some_peer() do
    case P2P.Peerbook.get_some_peer() do
      nil ->
        # TODO: (#1317) handle no-peers asynchronously
        raise "No peers available to request blocks from."

      peer_id ->
        peer_id
    end
  end

  defp parse_reason(reason) do
    case reason do
      "failed to dial" <> _ -> "failed to dial"
      res -> res
    end
  end

  defp verify_batch(blocks, start_slot, count) do
    end_slot = start_slot + count

    if Enum.all?(blocks, fn %{message: %{slot: slot}} ->
         start_slot <= slot and slot < end_slot
       end) do
      :ok
    else
      {:error, "block outside requested slot range"}
    end
  end

  defp flatten_response({:ok, [block]}), do: {:ok, block}
  defp flatten_response(other), do: other
end
