feat: add eth_getLogs rpc endpoint

CHANGELOG.md

@@ -4,6 +4,7 @@
 - [#2109](https://github.com/poanetwork/blockscout/pull/2109) - use bigger updates instead of `Multi` transactions in BlocksTransactionsMismatch
 - [#2075](https://github.com/poanetwork/blockscout/pull/2075) - add blocks cache
 - [#2151](https://github.com/poanetwork/blockscout/pull/2151) - hide dropdown menu then other networks list is empty
+- [#2146](https://github.com/poanetwork/blockscout/pull/2146) - feat: add eth_getLogs rpc endpoint
 
 ### Fixes
 - [#2162](https://github.com/poanetwork/blockscout/pull/2162) - contract creation tile color changed

apps/block_scout_web/lib/block_scout_web/controllers/api/rpc/eth_controller.ex

@@ -1,8 +1,33 @@
 defmodule BlockScoutWeb.API.RPC.EthController do
   use BlockScoutWeb, :controller
 
-  alias Explorer.Chain
-  alias Explorer.Chain.Wei
+  alias Ecto.Type, as: EctoType
+  alias Explorer.{Chain, Repo}
+  alias Explorer.Chain.{Block, Data, Hash, Hash.Address, Wei}
+  alias Explorer.Etherscan.Logs
+
+  @methods %{
+    "eth_getBalance" => %{
+      action: :eth_get_balance,
+      notes: """
+      the `earliest` parameter will not work as expected currently, because genesis block balances
+      are not currently imported
+      """
+    },
+    "eth_getLogs" => %{
+      action: :eth_get_logs,
+      notes: """
+      Will never return more than 1000 log entries.
+      """
+    }
+  }
+
+  @index_to_word %{
+    0 => "first",
+    1 => "second",
+    2 => "third",
+    3 => "fourth"
+  }
 
   def eth_request(%{body_params: %{"_json" => requests}} = conn, _) when is_list(requests) do
     responses = responses(requests)
@@ -39,6 +64,138 @@ defmodule BlockScoutWeb.API.RPC.EthController do
     |> render("response.json", %{response: response})
   end
 
+  def eth_get_balance(address_param, block_param \\ nil) do
+    with {:address, {:ok, address}} <- {:address, Chain.string_to_address_hash(address_param)},
+         {:block, {:ok, block}} <- {:block, block_param(block_param)},
+         {:balance, {:ok, balance}} <- {:balance, Chain.get_balance_as_of_block(address, block)} do
+      {:ok, Wei.hex_format(balance)}
+    else
+      {:address, :error} ->
+        {:error, "Query parameter 'address' is invalid"}
+
+      {:block, :error} ->
+        {:error, "Query parameter 'block' is invalid"}
+
+      {:balance, {:error, :not_found}} ->
+        {:error, "Balance not found"}
+    end
+  end
+
+  def eth_get_logs(filter_options) do
+    with {:ok, address_or_topic_params} <- address_or_topic_params(filter_options),
+         {:ok, from_block_param, to_block_param} <- logs_blocks_filter(filter_options),
+         {:ok, from_block} <- cast_block(from_block_param),
+         {:ok, to_block} <- cast_block(to_block_param) do
+      filter =
+        address_or_topic_params
+        |> Map.put(:from_block, from_block)
+        |> Map.put(:to_block, to_block)
+        |> Map.put(:allow_non_consensus, true)
+
+      {:ok, filter |> Logs.list_logs() |> Enum.map(&render_log/1)}
+    else
+      {:error, message} when is_bitstring(message) ->
+        {:error, message}
+
+      {:error, :empty} ->
+        {:ok, []}
+
+      _ ->
+        {:error, "Something went wrong."}
+    end
+  end
+
+  defp render_log(log) do
+    topics = Enum.reject([log.first_topic, log.second_topic, log.third_topic, log.fourth_topic], &is_nil/1)
+
+    %{
+      "address" => to_string(log.address_hash),
+      "blockHash" => to_string(log.block_hash),
+      "blockNumber" => Integer.to_string(log.block_number, 16),
+      "data" => to_string(log.data),
+      "logIndex" => Integer.to_string(log.index, 16),
+      "removed" => log.block_consensus == false,
+      "topics" => topics,
+      "transactionHash" => to_string(log.transaction_hash),
+      "transactionIndex" => log.transaction_index,
+      "transactionLogIndex" => log.index,
+      "type" => "mined"
+    }
+  end
+
+  defp cast_block("0x" <> hexadecimal_digits = input) do
+    case Integer.parse(hexadecimal_digits, 16) do
+      {integer, ""} -> {:ok, integer}
+      _ -> {:error, input <> " is not a valid block number"}
+    end
+  end
+
+  defp cast_block(integer) when is_integer(integer), do: {:ok, integer}
+  defp cast_block(_), do: {:error, "invalid block number"}
+
+  defp address_or_topic_params(filter_options) do
+    address_param = Map.get(filter_options, "address")
+    topics_param = Map.get(filter_options, "topics")
+
+    with {:ok, address} <- validate_address(address_param),
+         {:ok, topics} <- validate_topics(topics_param) do
+      address_and_topics(address, topics)
+    end
+  end
+
+  defp address_and_topics(nil, nil), do: {:error, "Must supply one of address and topics"}
+  defp address_and_topics(address, nil), do: {:ok, %{address_hash: address}}
+  defp address_and_topics(nil, topics), do: {:ok, topics}
+  defp address_and_topics(address, topics), do: {:ok, Map.put(topics, :address_hash, address)}
+
+  defp validate_address(nil), do: {:ok, nil}
+
+  defp validate_address(address) do
+    case Address.cast(address) do
+      {:ok, address} -> {:ok, address}
+      :error -> {:error, "invalid address"}
+    end
+  end
+
+  defp validate_topics(nil), do: {:ok, nil}
+  defp validate_topics([]), do: []
+
+  defp validate_topics(topics) when is_list(topics) do
+    topics
+    |> Stream.with_index()
+    |> Enum.reduce({:ok, %{}}, fn {topic, index}, {:ok, acc} ->
+      case cast_topics(topic) do
+        {:ok, data} ->
+          with_filter = Map.put(acc, String.to_existing_atom("#{@index_to_word[index]}_topic"), data)
+
+          {:ok, add_operator(with_filter, index)}
+
+        :error ->
+          {:error, "invalid topics"}
+      end
+    end)
+  end
+
+  defp add_operator(filters, 0), do: filters
+
+  defp add_operator(filters, index) do
+    Map.put(filters, String.to_existing_atom("topic#{index - 1}_#{index}_opr"), "and")
+  end
+
+  defp cast_topics(topics) when is_list(topics) do
+    case EctoType.cast({:array, Data}, topics) do
+      {:ok, data} -> {:ok, Enum.map(data, &to_string/1)}
+      :error -> :error
+    end
+  end
+
+  defp cast_topics(topic) do
+    case Data.cast(topic) do
+      {:ok, data} -> {:ok, to_string(data)}
+      :error -> :error
+    end
+  end
+
   defp responses(requests) do
     Enum.map(requests, fn request ->
       with {:id, {:ok, id}} <- {:id, Map.fetch(request, "id")},
@@ -51,6 +208,85 @@ defmodule BlockScoutWeb.API.RPC.EthController do
     end)
   end
 
+  defp logs_blocks_filter(filter_options) do
+    with {:filter, %{"blockHash" => block_hash_param}} <- {:filter, filter_options},
+         {:block_hash, {:ok, block_hash}} <- {:block_hash, Hash.Full.cast(block_hash_param)},
+         {:block, %{number: number}} <- {:block, Repo.get(Block, block_hash)} do
+      {:ok, number, number}
+    else
+      {:filter, filters} ->
+        from_block = Map.get(filters, "fromBlock", "latest")
+        to_block = Map.get(filters, "toBlock", "latest")
+
+        max_block_number =
+          if from_block == "latest" || to_block == "latest" do
+            max_consensus_block_number()
+          end
+
+        pending_block_number =
+          if from_block == "pending" || to_block == "pending" do
+            max_non_consensus_block_number(max_block_number)
+          end
+
+        if is_nil(pending_block_number) && from_block == "pending" && to_block == "pending" do
+          {:error, :empty}
+        else
+          to_block_numbers(from_block, to_block, max_block_number, pending_block_number)
+        end
+
+      {:block, _} ->
+        {:error, "Invalid Block Hash"}
+
+      {:block_hash, _} ->
+        {:error, "Invalid Block Hash"}
+    end
+  end
+
+  defp to_block_numbers(from_block, to_block, max_block_number, pending_block_number) do
+    actual_pending_block_number = pending_block_number || max_block_number
+
+    with {:ok, from} <- to_block_number(from_block, max_block_number, actual_pending_block_number),
+         {:ok, to} <- to_block_number(to_block, max_block_number, actual_pending_block_number) do
+      {:ok, from, to}
+    end
+  end
+
+  defp to_block_number(integer, _, _) when is_integer(integer), do: {:ok, integer}
+  defp to_block_number("latest", max_block_number, _), do: {:ok, max_block_number || 0}
+  defp to_block_number("earliest", _, _), do: {:ok, 0}
+  defp to_block_number("pending", max_block_number, nil), do: {:ok, max_block_number || 0}
+  defp to_block_number("pending", _, pending), do: {:ok, pending}
+
+  defp to_block_number("0x" <> number, _, _) do
+    case Integer.parse(number, 16) do
+      {integer, ""} -> {:ok, integer}
+      _ -> {:error, "invalid block number"}
+    end
+  end
+
+  defp to_block_number(number, _, _) when is_bitstring(number) do
+    case Integer.parse(number, 16) do
+      {integer, ""} -> {:ok, integer}
+      _ -> {:error, "invalid block number"}
+    end
+  end
+
+  defp to_block_number(_, _, _), do: {:error, "invalid block number"}
+
+  defp max_non_consensus_block_number(max) do
+    case Chain.max_non_consensus_block_number(max) do
+      {:ok, number} -> number
+      _ -> nil
+    end
+  end
+
+  defp max_consensus_block_number do
+    case Chain.max_consensus_block_number() do
+      {:ok, number} -> number
+      _ -> nil
+    end
+  end
+
   defp format_success(result, id) do
     %{result: result, id: id}
   end
@@ -66,9 +302,13 @@ defmodule BlockScoutWeb.API.RPC.EthController do
   defp do_eth_request(%{"jsonrpc" => "2.0", "method" => method, "params" => params})
        when is_list(params) do
     with {:ok, action} <- get_action(method),
-         true <- :erlang.function_exported(__MODULE__, action, Enum.count(params)) do
+         {:correct_arity, true} <-
+           {:correct_arity, :erlang.function_exported(__MODULE__, action, Enum.count(params))} do
       apply(__MODULE__, action, params)
     else
+      {:correct_arity, _} ->
+        {:error, "Incorrect number of params."}
+
       _ ->
         {:error, "Action not found."}
     end
@@ -82,26 +322,16 @@ defmodule BlockScoutWeb.API.RPC.EthController do
     {:error, "Method, params, and jsonrpc, are all required parameters."}
   end
 
-  def eth_get_balance(address_param, block_param \\ nil) do
-    with {:address, {:ok, address}} <- {:address, Chain.string_to_address_hash(address_param)},
-         {:block, {:ok, block}} <- {:block, block_param(block_param)},
-         {:balance, {:ok, balance}} <- {:balance, Chain.get_balance_as_of_block(address, block)} do
-      {:ok, Wei.hex_format(balance)}
-    else
-      {:address, :error} ->
-        {:error, "Query parameter 'address' is invalid"}
-
-      {:block, :error} ->
-        {:error, "Query parameter 'block' is invalid"}
+  defp get_action(action) do
+    case Map.get(@methods, action) do
+      %{action: action} ->
+        {:ok, action}
 
-      {:balance, {:error, :not_found}} ->
-        {:error, "Balance not found"}
+      _ ->
+        :error
     end
   end
 
-  defp get_action("eth_getBalance"), do: {:ok, :eth_get_balance}
-  defp get_action(_), do: :error
-
   defp block_param("latest"), do: {:ok, :latest}
   defp block_param("earliest"), do: {:ok, :earliest}
   defp block_param("pending"), do: {:ok, :pending}

apps/block_scout_web/test/block_scout_web/controllers/api/rpc/eth_controller_test.exs

@@ -2,6 +2,7 @@ defmodule BlockScoutWeb.API.RPC.EthControllerTest do
   use BlockScoutWeb.ConnCase, async: false
 
   alias Explorer.Counters.{AddressesWithBalanceCounter, AverageBlockTime}
+  alias Explorer.Repo
   alias Indexer.Fetcher.CoinBalanceOnDemand
 
   setup do
@@ -26,6 +27,291 @@ defmodule BlockScoutWeb.API.RPC.EthControllerTest do
 
   defp params(api_params, params), do: Map.put(api_params, "params", params)
 
+  describe "eth_get_logs" do
+    setup do
+      %{
+        api_params: %{
+          "method" => "eth_getLogs",
+          "jsonrpc" => "2.0",
+          "id" => 0
+        }
+      }
+    end
+
+    test "with an invalid address", %{conn: conn, api_params: api_params} do
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params(api_params, [%{"address" => "badhash"}]))
+               |> json_response(200)
+
+      assert %{"error" => "invalid address"} = response
+    end
+
+    test "address with no logs", %{conn: conn, api_params: api_params} do
+      insert(:block)
+      address = insert(:address)
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params(api_params, [%{"address" => to_string(address.hash)}]))
+               |> json_response(200)
+
+      assert %{"result" => []} = response
+    end
+
+    test "address but no logs and no toBlock provided", %{conn: conn, api_params: api_params} do
+      address = insert(:address)
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params(api_params, [%{"address" => to_string(address.hash)}]))
+               |> json_response(200)
+
+      assert %{"result" => []} = response
+    end
+
+    test "with a matching address", %{conn: conn, api_params: api_params} do
+      address = insert(:address)
+
+      block = insert(:block, number: 0)
+
+      transaction = insert(:transaction, from_address: address) |> with_block(block)
+      insert(:log, address: address, transaction: transaction, data: "0x010101")
+
+      params = params(api_params, [%{"address" => to_string(address.hash)}])
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params)
+               |> json_response(200)
+
+      assert %{"result" => [%{"data" => "0x010101"}]} = response
+    end
+
+    test "with a matching address and matching topic", %{conn: conn, api_params: api_params} do
+      address = insert(:address)
+
+      block = insert(:block, number: 0)
+
+      transaction = insert(:transaction, from_address: address) |> with_block(block)
+      insert(:log, address: address, transaction: transaction, data: "0x010101", first_topic: "0x01")
+
+      params = params(api_params, [%{"address" => to_string(address.hash), "topics" => ["0x01"]}])
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params)
+               |> json_response(200)
+
+      assert %{"result" => [%{"data" => "0x010101"}]} = response
+    end
+
+    test "with a matching address and multiple topic matches", %{conn: conn, api_params: api_params} do
+      address = insert(:address)
+
+      block = insert(:block, number: 0)
+
+      transaction = insert(:transaction, from_address: address) |> with_block(block)
+      insert(:log, address: address, transaction: transaction, data: "0x010101", first_topic: "0x01")
+      insert(:log, address: address, transaction: transaction, data: "0x020202", first_topic: "0x00")
+
+      params = params(api_params, [%{"address" => to_string(address.hash), "topics" => [["0x01", "0x00"]]}])
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params)
+               |> json_response(200)
+
+      assert [%{"data" => "0x010101"}, %{"data" => "0x020202"}] = Enum.sort_by(response["result"], &Map.get(&1, "data"))
+    end
+
+    test "with a matching address and multiple topic matches in different positions", %{
+      conn: conn,
+      api_params: api_params
+    } do
+      address = insert(:address)
+
+      block = insert(:block, number: 0)
+
+      transaction = insert(:transaction, from_address: address) |> with_block(block)
+
+      insert(:log,
+        address: address,
+        transaction: transaction,
+        data: "0x010101",
+        first_topic: "0x01",
+        second_topic: "0x02"
+      )
+
+      insert(:log, address: address, transaction: transaction, data: "0x020202", first_topic: "0x01")
+
+      params = params(api_params, [%{"address" => to_string(address.hash), "topics" => ["0x01", "0x02"]}])
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params)
+               |> json_response(200)
+
+      assert [%{"data" => "0x010101"}] = response["result"]
+    end
+
+    test "with a matching address and multiple topic matches in different positions and multiple matches in the second position",
+         %{conn: conn, api_params: api_params} do
+      address = insert(:address)
+
+      block = insert(:block, number: 0)
+
+      transaction = insert(:transaction, from_address: address) |> with_block(block)
+
+      insert(:log,
+        address: address,
+        transaction: transaction,
+        data: "0x010101",
+        first_topic: "0x01",
+        second_topic: "0x02"
+      )
+
+      insert(:log,
+        address: address,
+        transaction: transaction,
+        data: "0x020202",
+        first_topic: "0x01",
+        second_topic: "0x03"
+      )
+
+      params = params(api_params, [%{"address" => to_string(address.hash), "topics" => ["0x01", ["0x02", "0x03"]]}])
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params)
+               |> json_response(200)
+
+      assert [%{"data" => "0x010101"}, %{"data" => "0x020202"}] = Enum.sort_by(response["result"], &Map.get(&1, "data"))
+    end
+
+    test "with a block range filter",
+         %{conn: conn, api_params: api_params} do
+      address = insert(:address)
+
+      block1 = insert(:block, number: 0)
+      block2 = insert(:block, number: 1)
+      block3 = insert(:block, number: 2)
+      block4 = insert(:block, number: 3)
+
+      transaction1 = insert(:transaction, from_address: address) |> with_block(block1)
+      transaction2 = insert(:transaction, from_address: address) |> with_block(block2)
+      transaction3 = insert(:transaction, from_address: address) |> with_block(block3)
+      transaction4 = insert(:transaction, from_address: address) |> with_block(block4)
+
+      insert(:log, address: address, transaction: transaction1, data: "0x010101")
+
+      insert(:log, address: address, transaction: transaction2, data: "0x020202")
+
+      insert(:log, address: address, transaction: transaction3, data: "0x030303")
+
+      insert(:log, address: address, transaction: transaction4, data: "0x040404")
+
+      params = params(api_params, [%{"address" => to_string(address.hash), "fromBlock" => 1, "toBlock" => 2}])
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params)
+               |> json_response(200)
+
+      assert [%{"data" => "0x020202"}, %{"data" => "0x030303"}] = Enum.sort_by(response["result"], &Map.get(&1, "data"))
+    end
+
+    test "with a block hash filter",
+         %{conn: conn, api_params: api_params} do
+      address = insert(:address)
+
+      block1 = insert(:block, number: 0)
+      block2 = insert(:block, number: 1)
+      block3 = insert(:block, number: 2)
+
+      transaction1 = insert(:transaction, from_address: address) |> with_block(block1)
+      transaction2 = insert(:transaction, from_address: address) |> with_block(block2)
+      transaction3 = insert(:transaction, from_address: address) |> with_block(block3)
+
+      insert(:log, address: address, transaction: transaction1, data: "0x010101")
+
+      insert(:log, address: address, transaction: transaction2, data: "0x020202")
+
+      insert(:log, address: address, transaction: transaction3, data: "0x030303")
+
+      params = params(api_params, [%{"address" => to_string(address.hash), "blockHash" => to_string(block2.hash)}])
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params)
+               |> json_response(200)
+
+      assert [%{"data" => "0x020202"}] = response["result"]
+    end
+
+    test "with an earliest block filter",
+         %{conn: conn, api_params: api_params} do
+      address = insert(:address)
+
+      block1 = insert(:block, number: 0)
+      block2 = insert(:block, number: 1)
+      block3 = insert(:block, number: 2)
+
+      transaction1 = insert(:transaction, from_address: address) |> with_block(block1)
+      transaction2 = insert(:transaction, from_address: address) |> with_block(block2)
+      transaction3 = insert(:transaction, from_address: address) |> with_block(block3)
+
+      insert(:log, address: address, transaction: transaction1, data: "0x010101")
+
+      insert(:log, address: address, transaction: transaction2, data: "0x020202")
+
+      insert(:log, address: address, transaction: transaction3, data: "0x030303")
+
+      params =
+        params(api_params, [%{"address" => to_string(address.hash), "fromBlock" => "earliest", "toBlock" => "earliest"}])
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params)
+               |> json_response(200)
+
+      assert [%{"data" => "0x010101"}] = response["result"]
+    end
+
+    test "with a pending block filter",
+         %{conn: conn, api_params: api_params} do
+      address = insert(:address)
+
+      block1 = insert(:block, number: 0)
+      block2 = insert(:block, number: 1)
+      block3 = insert(:block, number: 2)
+
+      transaction1 = insert(:transaction, from_address: address) |> with_block(block1)
+      transaction2 = insert(:transaction, from_address: address) |> with_block(block2)
+      transaction3 = insert(:transaction, from_address: address) |> with_block(block3)
+
+      insert(:log, address: address, transaction: transaction1, data: "0x010101")
+
+      insert(:log, address: address, transaction: transaction2, data: "0x020202")
+
+      insert(:log, address: address, transaction: transaction3, data: "0x030303")
+
+      changeset = Ecto.Changeset.change(block3, %{consensus: false})
+
+      Repo.update!(changeset)
+
+      params =
+        params(api_params, [%{"address" => to_string(address.hash), "fromBlock" => "pending", "toBlock" => "pending"}])
+
+      assert response =
+               conn
+               |> post("/api/eth_rpc", params)
+               |> json_response(200)
+
+      assert [%{"data" => "0x030303"}] = response["result"]
+    end
+  end
+
   describe "eth_get_balance" do
     setup do
       %{

apps/explorer/lib/explorer/chain.ex

@@ -1700,6 +1700,32 @@ defmodule Explorer.Chain do
     end
   end
 
+  @spec max_non_consensus_block_number(integer | nil) :: {:ok, Block.block_number()} | {:error, :not_found}
+  def max_non_consensus_block_number(max_consensus_block_number \\ nil) do
+    max =
+      if max_consensus_block_number do
+        {:ok, max_consensus_block_number}
+      else
+        max_consensus_block_number()
+      end
+
+    case max do
+      {:ok, number} ->
+        query =
+          from(block in Block,
+            where: block.consensus == false,
+            where: block.number > ^number
+          )
+
+        query
+        |> Repo.aggregate(:max, :number)
+        |> case do
+          nil -> {:error, :not_found}
+          number -> {:ok, number}
+        end
+    end
+  end
+
   @doc """
   The height of the chain.
 

apps/explorer/lib/explorer/etherscan/logs.ex

@@ -34,7 +34,8 @@ defmodule Explorer.Etherscan.Logs do
     :fourth_topic,
     :index,
     :address_hash,
-    :transaction_hash
+    :transaction_hash,
+    :type
   ]
 
   @doc """
@@ -114,16 +115,26 @@ defmodule Explorer.Etherscan.Logs do
       from(log_transaction_data in subquery(all_transaction_logs_query),
         join: block in Block,
         on: block.number == log_transaction_data.block_number,
-        where: block.consensus == true,
         where: log_transaction_data.address_hash == ^address_hash,
         order_by: block.number,
         limit: 1000,
         select_merge: %{
-          block_timestamp: block.timestamp
+          block_timestamp: block.timestamp,
+          block_consensus: block.consensus,
+          block_hash: block.hash
         }
       )
 
-    Repo.all(query_with_blocks)
+    query_with_consensus =
+      if Map.get(filter, :allow_non_consensus) do
+        query_with_blocks
+      else
+        from([_, block] in query_with_blocks,
+          where: block.consensus == true
+        )
+      end
+
+    Repo.all(query_with_consensus)
   end
 
   # Since address_hash was not present, we know that a
@@ -140,20 +151,30 @@ defmodule Explorer.Etherscan.Logs do
         join: block in assoc(transaction, :block),
         where: block.number >= ^prepared_filter.from_block,
         where: block.number <= ^prepared_filter.to_block,
-        where: block.consensus == true,
         select: %{
           transaction_hash: transaction.hash,
           gas_price: transaction.gas_price,
           gas_used: transaction.gas_used,
           transaction_index: transaction.index,
+          block_hash: block.hash,
           block_number: block.number,
-          block_timestamp: block.timestamp
+          block_timestamp: block.timestamp,
+          block_consensus: block.consensus
         }
       )
 
+    query_with_consensus =
+      if Map.get(filter, :allow_non_consensus) do
+        block_transaction_query
+      else
+        from([_, block] in block_transaction_query,
+          where: block.consensus == true
+        )
+      end
+
     query_with_block_transaction_data =
       from(log in logs_query,
-        join: block_transaction_data in subquery(block_transaction_query),
+        join: block_transaction_data in subquery(query_with_consensus),
         on: block_transaction_data.transaction_hash == log.transaction_hash,
         order_by: block_transaction_data.block_number,
         limit: 1000,
@@ -186,7 +207,7 @@ defmodule Explorer.Etherscan.Logs do
         query
 
       [topic] ->
-        where(query, [l], field(l, ^topic) == ^filter[topic])
+        where(query, [l], field(l, ^topic) in ^List.wrap(filter[topic]))
 
       _ ->
         where_multiple_topics_match(query, filter)
@@ -201,12 +222,12 @@ defmodule Explorer.Etherscan.Logs do
 
   defp where_multiple_topics_match(query, filter, topic_operation, "and") do
     {topic_a, topic_b} = @topic_operations[topic_operation]
-    where(query, [l], field(l, ^topic_a) == ^filter[topic_a] and field(l, ^topic_b) == ^filter[topic_b])
+    where(query, [l], field(l, ^topic_a) == ^filter[topic_a] and field(l, ^topic_b) in ^List.wrap(filter[topic_b]))
   end
 
   defp where_multiple_topics_match(query, filter, topic_operation, "or") do
     {topic_a, topic_b} = @topic_operations[topic_operation]
-    where(query, [l], field(l, ^topic_a) == ^filter[topic_a] or field(l, ^topic_b) == ^filter[topic_b])
+    where(query, [l], field(l, ^topic_a) == ^filter[topic_a] or field(l, ^topic_b) in ^List.wrap(filter[topic_b]))
   end
 
   defp where_multiple_topics_match(query, _, _, _), do: query