Merge pull request #1095 from poanetwork/wsa-async-task-to-check-token-metadata

Async task to check token metadata

apps/explorer/lib/explorer/chain.ex

@@ -1939,6 +1939,26 @@ defmodule Explorer.Chain do
     )
   end
 
+  @doc """
+  Streams a list of token contract addresses that have been cataloged.
+  """
+  @spec stream_cataloged_token_contract_address_hashes(
+          initial :: accumulator,
+          reducer :: (entry :: Hash.Address.t(), accumulator -> accumulator)
+        ) :: {:ok, accumulator}
+        when accumulator: term()
+  def stream_cataloged_token_contract_address_hashes(initial_acc, reducer) when is_function(reducer, 2) do
+    Repo.transaction(
+      fn ->
+        Chain.Token.cataloged_tokens()
+        |> order_by(asc: :updated_at)
+        |> Repo.stream(timeout: :infinity)
+        |> Enum.reduce(initial_acc, reducer)
+      end,
+      timeout: :infinity
+    )
+  end
+
   @doc """
   Returns a list of block numbers token transfer `t:Log.t/0`s that don't have an
   associated `t:TokenTransfer.t/0` record.

apps/explorer/lib/explorer/chain/token.ex

@@ -94,4 +94,17 @@ defmodule Explorer.Chain.Token do
       on: tt.token_contract_address_hash == t.contract_address_hash
     )
   end
+
+  @doc """
+  Builds an `Ecto.Query` to fetch the cataloged tokens.
+
+  These are tokens with cataloged field set to true.
+  """
+  def cataloged_tokens do
+    from(
+      token in __MODULE__,
+      select: token.contract_address_hash,
+      where: token.cataloged == true
+    )
+  end
 end

apps/explorer/test/explorer/chain/token_test.exs

@@ -0,0 +1,16 @@
+defmodule Explorer.Chain.TokenTest do
+  use Explorer.DataCase
+
+  import Explorer.Factory
+
+  alias Explorer.Chain
+
+  describe "cataloged_tokens/0" do
+    test "filters only cataloged tokens" do
+      token = insert(:token, cataloged: true)
+      insert(:token, cataloged: false)
+
+      assert Repo.all(Chain.Token.cataloged_tokens()) == [token.contract_address_hash]
+    end
+  end
+end

apps/explorer/test/explorer/chain_test.exs

@@ -2756,6 +2756,29 @@ defmodule Explorer.ChainTest do
     assert Chain.stream_uncataloged_token_contract_address_hashes([], &[&1 | &2]) == {:ok, [uncatalog_address]}
   end
 
+  describe "stream_cataloged_token_contract_address_hashes/2" do
+    test "reduces with given reducer and accumulator" do
+      %Token{contract_address_hash: catalog_address} = insert(:token, cataloged: true)
+      insert(:token, cataloged: false)
+      assert Chain.stream_cataloged_token_contract_address_hashes([], &[&1 | &2]) == {:ok, [catalog_address]}
+    end
+
+    test "sorts the tokens by updated_at in ascending order" do
+      today = DateTime.utc_now()
+      yesterday = Timex.shift(today, days: -1)
+
+      token1 = insert(:token, %{cataloged: true, updated_at: today})
+      token2 = insert(:token, %{cataloged: true, updated_at: yesterday})
+
+      expected_response =
+        [token1, token2]
+        |> Enum.sort(&(&1.updated_at < &2.updated_at))
+        |> Enum.map(& &1.contract_address_hash)
+
+      assert Chain.stream_cataloged_token_contract_address_hashes([], &(&2 ++ [&1])) == {:ok, expected_response}
+    end
+  end
+
   describe "transaction_has_token_transfers?/1" do
     test "returns true if transaction has token transfers" do
       transaction = insert(:transaction)

apps/indexer/config/config.exs

@@ -7,6 +7,7 @@ import Bitwise
 config :indexer,
   block_transformer: Indexer.Block.Transform.Base,
   ecto_repos: [Explorer.Repo],
+  metadata_updater_days_interval: 7,
   # bytes
   memory_limit: 1 <<< 30
 

apps/indexer/lib/indexer/token/metadata_updater.ex

@@ -0,0 +1,51 @@
+defmodule Indexer.Token.MetadataUpdater do
+  @moduledoc """
+  Updates metadata for cataloged tokens
+  """
+
+  use GenServer
+
+  alias Explorer.Chain
+  alias Explorer.Chain.Token
+  alias Explorer.Token.MetadataRetriever
+
+  def start_link(initial_state) do
+    GenServer.start_link(__MODULE__, initial_state, name: __MODULE__)
+  end
+
+  @impl true
+  def init(state) do
+    send(self(), :update_tokens)
+
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_info(:update_tokens, state) do
+    {:ok, tokens} = Chain.stream_cataloged_token_contract_address_hashes([], &[&1 | &2])
+
+    tokens
+    |> Enum.reverse()
+    |> update_metadata()
+
+    Process.send_after(self(), :update_tokens, :timer.hours(state.update_interval) * 24)
+
+    {:noreply, state}
+  end
+
+  @doc false
+  def update_metadata(token_addresses) when is_list(token_addresses) do
+    Enum.each(token_addresses, fn address ->
+      case Chain.token_from_address_hash(address) do
+        {:ok, %Token{cataloged: true} = token} ->
+          update_metadata(token)
+      end
+    end)
+  end
+
+  def update_metadata(%Token{contract_address_hash: contract_address_hash} = token) do
+    contract_functions = MetadataRetriever.get_functions_of(contract_address_hash)
+
+    Chain.update_token(%{token | updated_at: DateTime.utc_now()}, contract_functions)
+  end
+end

apps/indexer/lib/indexer/token/supervisor.ex

@@ -5,7 +5,7 @@ defmodule Indexer.Token.Supervisor do
 
   use Supervisor
 
-  alias Indexer.Token.Fetcher
+  alias Indexer.Token.{Fetcher, MetadataUpdater}
 
   def child_spec([init_arguments]) do
     child_spec([init_arguments, []])
@@ -27,10 +27,13 @@ defmodule Indexer.Token.Supervisor do
 
   @impl Supervisor
   def init(fetcher_arguments) do
+    metadata_updater_inverval = Application.get_env(:indexer, :metadata_updater_days_interval)
+
     Supervisor.init(
       [
         {Task.Supervisor, name: Indexer.Token.TaskSupervisor},
-        {Fetcher, [fetcher_arguments, [name: Fetcher]]}
+        {Fetcher, [fetcher_arguments, [name: Fetcher]]},
+        {MetadataUpdater, %{update_interval: metadata_updater_inverval}}
       ],
       strategy: :one_for_one
     )

apps/indexer/test/indexer/token/metadata_updater_test.exs

@@ -0,0 +1,107 @@
+defmodule Indexer.Token.MetadataUpdaterTest do
+  use Explorer.DataCase
+
+  import Mox
+
+  alias Explorer.Chain
+  alias Explorer.Chain.Token
+  alias Indexer.Token.MetadataUpdater
+
+  setup :verify_on_exit!
+  setup :set_mox_global
+
+  test "updates tokens metadata on start" do
+    insert(:token, name: nil, symbol: nil, decimals: 10, cataloged: true)
+
+    expect(
+      EthereumJSONRPC.Mox,
+      :json_rpc,
+      1,
+      fn [%{id: "decimals"}, %{id: "name"}, %{id: "symbol"}, %{id: "totalSupply"}], _opts ->
+        {:ok,
+         [
+           %{
+             id: "decimals",
+             result: "0x0000000000000000000000000000000000000000000000000000000000000012"
+           },
+           %{
+             id: "name",
+             result:
+               "0x0000000000000000000000000000000000000000000000000000000000000020000000000000000000000000000000000000000000000000000000000000000642616e636f720000000000000000000000000000000000000000000000000000"
+           },
+           %{
+             id: "symbol",
+             result:
+               "0x00000000000000000000000000000000000000000000000000000000000000200000000000000000000000000000000000000000000000000000000000000003424e540000000000000000000000000000000000000000000000000000000000"
+           },
+           %{
+             id: "totalSupply",
+             result: "0x0000000000000000000000000000000000000000000000000de0b6b3a7640000"
+           }
+         ]}
+      end
+    )
+
+    pid = start_supervised!({MetadataUpdater, %{update_interval: 0}})
+
+    wait_for_results(fn ->
+      updated = Repo.one!(from(t in Token, where: t.cataloged == true and not is_nil(t.name), limit: 1))
+
+      assert updated.name != nil
+      assert updated.symbol != nil
+    end)
+
+    # Terminates the process so it finishes all Ecto processes.
+    GenServer.stop(pid)
+  end
+
+  describe "update_metadata/1" do
+    test "updates the metadata for a list of tokens" do
+      token = insert(:token, name: nil, symbol: nil, decimals: 10)
+
+      expect(
+        EthereumJSONRPC.Mox,
+        :json_rpc,
+        1,
+        fn [%{id: "decimals"}, %{id: "name"}, %{id: "symbol"}, %{id: "totalSupply"}], _opts ->
+          {:ok,
+           [
+             %{
+               id: "decimals",
+               result: "0x0000000000000000000000000000000000000000000000000000000000000012"
+             },
+             %{
+               id: "name",
+               result:
+                 "0x0000000000000000000000000000000000000000000000000000000000000020000000000000000000000000000000000000000000000000000000000000000642616e636f720000000000000000000000000000000000000000000000000000"
+             },
+             %{
+               id: "symbol",
+               result:
+                 "0x00000000000000000000000000000000000000000000000000000000000000200000000000000000000000000000000000000000000000000000000000000003424e540000000000000000000000000000000000000000000000000000000000"
+             },
+             %{
+               id: "totalSupply",
+               result: "0x0000000000000000000000000000000000000000000000000de0b6b3a7640000"
+             }
+           ]}
+        end
+      )
+
+      MetadataUpdater.update_metadata([token.contract_address_hash])
+
+      expected_supply = Decimal.new(1_000_000_000_000_000_000)
+
+      decimals_expected = Decimal.new(18)
+
+      assert {:ok,
+              %Token{
+                name: "Bancor",
+                symbol: "BNT",
+                total_supply: ^expected_supply,
+                decimals: ^decimals_expected,
+                cataloged: true
+              }} = Chain.token_from_address_hash(token.contract_address_hash)
+    end
+  end
+end