Catalog token transfers and tokens during indexing (#484)

* Fetch token information asynchronously during indexing
* Include token transfers imports in main importer
* Extract token transfers when importing blocks while indexing
* Move ecto to explorer deps
* Update References to the Indexer namespace
* Add token insertions with main importer
* Import token address/type during block indexing

apps/ethereum_jsonrpc/lib/ethereum_jsonrpc/encoder.ex

@@ -88,23 +88,24 @@ defmodule EthereumJSONRPC.Encoder do
   @doc """
   Given a result from the blockchain, and the function selector, returns the result decoded.
   """
-  @spec decode_result({map(), %ABI.FunctionSelector{}}) :: {String.t(), [String.t()]}
+  @spec decode_result({map(), %ABI.FunctionSelector{}}) ::
+          {String.t(), {:ok, any()} | {:error, String.t() | :invalid_data}}
   def decode_result({%{error: %{code: code, message: message}, id: id}, _selector}) do
-    {id, ["#{code} => #{message}"]}
+    {id, {:error, "(#{code}) #{message}"}}
   end
 
   def decode_result({%{id: id, result: result}, function_selector}) do
-    types_list = format_list_types(function_selector.returns)
+    types_list = List.wrap(function_selector.returns)
 
-    decoded_result =
+    [decoded_data] =
       result
       |> String.slice(2..-1)
       |> Base.decode16!(case: :lower)
       |> TypeDecoder.decode_raw(types_list)
 
-    {id, decoded_result}
+    {id, {:ok, decoded_data}}
+  rescue
+    MatchError ->
+      {id, {:error, :invalid_data}}
   end
-
-  defp format_list_types(:string), do: [{:array, :string, 1}]
-  defp format_list_types(return_types), do: List.wrap(return_types)
 end

apps/ethereum_jsonrpc/mix.exs

@@ -61,8 +61,6 @@ defmodule EthereumJsonrpc.MixProject do
       {:credo, "0.9.2", only: [:dev, :test], runtime: false},
       # Static Type Checking
       {:dialyxir, "~> 0.5", only: [:dev, :test], runtime: false},
-      # Casting Ethereum-native types to Elixir-native types
-      {:ecto, "~> 2.2"},
       # Code coverage
       {:excoveralls, "~> 0.8.1", only: [:test]},
       # JSONRPC HTTP Post calls
@@ -70,11 +68,11 @@ defmodule EthereumJsonrpc.MixProject do
       # Decode/Encode JSON for JSONRPC
       {:jason, "~> 1.0"},
       # Mocking `EthereumJSONRPC.Transport` and `EthereumJSONRPC.HTTP` so we avoid hitting real chains for local testing
-      {:mox, "~> 0.3.2", only: [:test]},
+      {:mox, "~> 0.4", only: [:test]},
       # Convert unix timestamps in JSONRPC to DateTimes
       {:timex, "~> 3.1.24"},
       # Encode/decode function names and arguments
-      {:ex_abi, "~> 0.1.13"}
+      {:ex_abi, "~> 0.1.14"}
     ]
   end
 end

apps/ethereum_jsonrpc/test/ethereum_jsonrpc/encoder_test.exs

@@ -151,9 +151,9 @@ defmodule EthereumJSONRPC.EncoderTest do
       }
 
       assert Encoder.decode_abi_results(result, abi, functions) == %{
-               "get1" => [42],
-               "get2" => [42],
-               "get3" => [32]
+               "get1" => {:ok, 42},
+               "get2" => {:ok, 42},
+               "get3" => {:ok, 32}
              }
     end
   end
@@ -172,7 +172,7 @@ defmodule EthereumJSONRPC.EncoderTest do
         types: [{:uint, 256}]
       }
 
-      assert Encoder.decode_result({result, selector}) == {"sum", [42]}
+      assert Encoder.decode_result({result, selector}) == {"sum", {:ok, 42}}
     end
 
     test "correclty handles the blockchain error response" do
@@ -192,7 +192,7 @@ defmodule EthereumJSONRPC.EncoderTest do
       }
 
       assert Encoder.decode_result({result, selector}) ==
-               {"sum", ["-32602 => Invalid params: Invalid hex: Invalid character 'x' at position 134."]}
+               {"sum", {:error, "(-32602) Invalid params: Invalid hex: Invalid character 'x' at position 134."}}
     end
 
     test "correclty decodes string types" do
@@ -201,7 +201,7 @@ defmodule EthereumJSONRPC.EncoderTest do
 
       selector = %ABI.FunctionSelector{function: "name", types: [], returns: :string}
 
-      assert Encoder.decode_result({%{id: "storedName", result: result}, selector}) == {"storedName", [["AION"]]}
+      assert Encoder.decode_result({%{id: "storedName", result: result}, selector}) == {"storedName", {:ok, "AION"}}
     end
   end
 end

apps/explorer/config/config.exs

@@ -18,8 +18,6 @@ config :explorer, Explorer.Chain.Statistics.Server, enabled: true
 
 config :explorer, Explorer.ExchangeRates, enabled: true
 
-config :explorer, Explorer.Indexer.Supervisor, enabled: true
-
 config :explorer, Explorer.Market.History.Cataloger, enabled: true
 
 config :explorer, Explorer.Repo, migration_timestamps: [type: :utc_datetime]

apps/explorer/config/test.exs

@@ -17,8 +17,6 @@ config :explorer, Explorer.Chain.Statistics.Server, enabled: false
 
 config :explorer, Explorer.ExchangeRates, enabled: false
 
-config :explorer, Explorer.Indexer.Supervisor, enabled: false
-
 config :explorer, Explorer.Market.History.Cataloger, enabled: false
 
 if File.exists?(file = "test.secret.exs") do

apps/explorer/lib/explorer/chain.ex

@@ -26,9 +26,10 @@ defmodule Explorer.Chain do
     Import,
     InternalTransaction,
     Log,
+    SmartContract,
+    Token,
     Transaction,
-    Wei,
-    SmartContract
+    Wei
   }
 
   alias Explorer.Chain.Block.Reward
@@ -1660,4 +1661,44 @@ defmodule Explorer.Chain do
   defp supply_module do
     Application.get_env(:explorer, :supply, Explorer.Chain.Supply.ProofOfAuthority)
   end
+
+  @doc """
+  Streams a lists token contract addresses that haven't been cataloged.
+  """
+  @spec stream_uncataloged_token_contract_address_hashes(
+          initial :: accumulator,
+          reducer :: (entry :: Hash.Address.t(), accumulator -> accumulator)
+        ) :: {:ok, accumulator}
+        when accumulator: term()
+  def stream_uncataloged_token_contract_address_hashes(initial_acc, reducer) when is_function(reducer, 2) do
+    Repo.transaction(
+      fn ->
+        query =
+          from(
+            token in Token,
+            where: token.cataloged == false,
+            select: token.contract_address_hash
+          )
+
+        query
+        |> Repo.stream(timeout: :infinity)
+        |> Enum.reduce(initial_acc, reducer)
+      end,
+      timeout: :infinity
+    )
+  end
+
+  @doc """
+  Fetches a `t:Token.t/0` by an address hash.
+  """
+  @spec token_from_address_hash(Hash.Address.t()) :: {:ok, Token.t()} | {:error, :not_found}
+  def token_from_address_hash(%Hash{byte_count: unquote(Hash.Address.byte_count())} = hash) do
+    case Repo.get_by(Token, contract_address_hash: hash) do
+      nil ->
+        {:error, :not_found}
+
+      %Token{} = token ->
+        {:ok, token}
+    end
+  end
 end

apps/explorer/lib/explorer/chain/import.ex

@@ -6,7 +6,20 @@ defmodule Explorer.Chain.Import do
   import Ecto.Query, only: [from: 2]
 
   alias Ecto.{Changeset, Multi}
-  alias Explorer.Chain.{Address, Balance, Block, Hash, InternalTransaction, Log, Transaction, Wei}
+
+  alias Explorer.Chain.{
+    Address,
+    Balance,
+    Block,
+    Hash,
+    InternalTransaction,
+    Log,
+    Token,
+    TokenTransfer,
+    Transaction,
+    Wei
+  }
+
   alias Explorer.Repo
 
   @type changeset_function_name :: atom
@@ -37,6 +50,15 @@ defmodule Explorer.Chain.Import do
           required(:params) => params,
           optional(:timeout) => timeout
         }
+  @type token_transfers_options :: %{
+          required(:params) => params,
+          optional(:timeout) => timeout
+        }
+  @type tokens_options :: %{
+          required(:params) => params,
+          optional(:on_conflict) => :nothing | :replace_all,
+          optional(:timeout) => timeout
+        }
   @type transactions_options :: %{
           required(:params) => params,
           optional(:with) => changeset_function_name,
@@ -53,6 +75,8 @@ defmodule Explorer.Chain.Import do
           optional(:logs) => logs_options,
           optional(:receipts) => receipts_options,
           optional(:timeout) => timeout,
+          optional(:token_transfers) => token_transfers_options,
+          optional(:tokens) => tokens_options,
           optional(:transactions) => transactions_options
         }
   @type all_result ::
@@ -68,6 +92,8 @@ defmodule Explorer.Chain.Import do
              ],
              optional(:logs) => [Log.t()],
              optional(:receipts) => [Hash.Full.t()],
+             optional(:token_transfers) => [TokenTransfer.t()],
+             optional(:tokens) => [Token.t()],
              optional(:transactions) => [Hash.Full.t()]
            }}
           | {:error, [Changeset.t()]}
@@ -85,6 +111,8 @@ defmodule Explorer.Chain.Import do
   @insert_blocks_timeout 60_000
   @insert_internal_transactions_timeout 60_000
   @insert_logs_timeout 60_000
+  @insert_token_transfers_timeout 60_000
+  @insert_tokens_timeout 60_000
   @insert_transactions_timeout 60_000
 
   @doc """
@@ -99,6 +127,8 @@ defmodule Explorer.Chain.Import do
   | `:blocks`                | `[Explorer.Chain.Block.t()]`                                                                    | List of `t:Explorer.Chain.Block.t/0`s                                                         |
   | `:internal_transactions` | `[%{index: non_neg_integer(), transaction_hash: Explorer.Chain.Hash.t()}]`                      | List of maps of the `t:Explorer.Chain.InternalTransaction.t/0` `index` and `transaction_hash` |
   | `:logs`                  | `[Explorer.Chain.Log.t()]`                                                                      | List of `t:Explorer.Chain.Log.t/0`s                                                           |
+  | `:token_transfers`       | `[Explorer.Chain.TokenTransfer.t()]`                                                            | List of `t:Explor.Chain.TokenTransfer.t/0`s                                                   |
+  | `:tokens`                | `[Explorer.Chain.Token.t()]`                                                                    | List of `t:Explorer.Chain.token.t/0`s                                                         |
   | `:transactions`          | `[Explorer.Chain.Hash.t()]`                                                                     | List of `t:Explorer.Chain.Transaction.t/0` `hash`                                             |
 
   The params for each key are validated using the corresponding `Ecto.Schema` module's `changeset/2` function.  If there
@@ -138,6 +168,14 @@ defmodule Explorer.Chain.Import do
       * `:timeout` - the timeout for inserting all logs. Defaults to `#{@insert_logs_timeout}` milliseconds.
     * `:timeout` - the timeout for the whole `c:Ecto.Repo.transaction/0` call.  Defaults to `#{@transaction_timeout}`
       milliseconds.
+    * `:token_transfers`
+      * `:params` - `list` of params for `Explorer.Chain.TokenTransfer.changeset/2`
+      * `:timeout` - the timeout for inserting all token transfers. Defaults to `#{@insert_token_transfers_timeout}` milliseconds.
+    * `:tokens`
+      * `:on_conflict` - Whether to do `:nothing` or `:replace_all` columns when there is a pre-existing token
+        with the same contract address hash.
+      * `:params` - `list` of params for `Explorer.Chain.Token.changeset/2`
+      * `:timeout` - the timeout for inserting all tokens. Defaults to `#{@insert_tokens_timeout}` milliseconds.
     * `:transactions`
       * `:on_conflict` - Whether to do `:nothing` or `:replace_all` columns when there is a pre-existing transaction
         with the same hash.
@@ -230,6 +268,8 @@ defmodule Explorer.Chain.Import do
     blocks: Block,
     internal_transactions: InternalTransaction,
     logs: Log,
+    token_transfers: TokenTransfer,
+    tokens: Token,
     transactions: Transaction
   }
 
@@ -245,6 +285,8 @@ defmodule Explorer.Chain.Import do
     |> run_transactions(ecto_schema_module_to_changes_list_map, full_options)
     |> run_internal_transactions(ecto_schema_module_to_changes_list_map, full_options)
     |> run_logs(ecto_schema_module_to_changes_list_map, full_options)
+    |> run_tokens(ecto_schema_module_to_changes_list_map, full_options)
+    |> run_token_transfers(ecto_schema_module_to_changes_list_map, full_options)
   end
 
   defp run_addresses(multi, ecto_schema_module_to_changes_list_map, options)
@@ -386,6 +428,51 @@ defmodule Explorer.Chain.Import do
     end
   end
 
+  defp run_tokens(multi, ecto_schema_module_to_changes_list, options)
+       when is_map(ecto_schema_module_to_changes_list) and is_map(options) do
+    case ecto_schema_module_to_changes_list do
+      %{Token => tokens_changes} ->
+        tokens_options = Map.fetch!(options, :tokens)
+        timestamps = Map.fetch!(options, :timestamps)
+        on_conflict = Map.fetch!(tokens_options, :on_conflict)
+
+        Multi.run(multi, :tokens, fn _ ->
+          insert_tokens(
+            tokens_changes,
+            %{
+              on_conflict: on_conflict,
+              timeout: options[:tokens][:timeout] || @insert_tokens_timeout,
+              timestamps: timestamps
+            }
+          )
+        end)
+
+      _ ->
+        multi
+    end
+  end
+
+  defp run_token_transfers(multi, ecto_schema_module_to_changes_list, options)
+       when is_map(ecto_schema_module_to_changes_list) and is_map(options) do
+    case ecto_schema_module_to_changes_list do
+      %{TokenTransfer => token_transfers_changes} ->
+        timestamps = Map.fetch!(options, :timestamps)
+
+        Multi.run(multi, :token_transfers, fn _ ->
+          insert_token_transfers(
+            token_transfers_changes,
+            %{
+              timeout: options[:token_transfers][:timeout] || @insert_token_transfers_timeout,
+              timestamps: timestamps
+            }
+          )
+        end)
+
+      _ ->
+        multi
+    end
+  end
+
   @spec insert_addresses([%{hash: Hash.Address.t()}], %{
           required(:timeout) => timeout,
           required(:timestamps) => timestamps
@@ -565,6 +652,50 @@ defmodule Explorer.Chain.Import do
       )
   end
 
+  @spec insert_tokens([map()], %{
+          required(:on_conflict) => on_conflict(),
+          required(:timeout) => timeout(),
+          required(:timestamps) => timestamps()
+        }) ::
+          {:ok, [Token.t()]}
+          | {:error, [Changeset.t()]}
+  def insert_tokens(changes_list, %{on_conflict: on_conflict, timeout: timeout, timestamps: timestamps})
+      when is_list(changes_list) do
+    # order so that row ShareLocks are grabbed in a consistent order
+    ordered_changes_list = Enum.sort_by(changes_list, & &1.contract_address_hash)
+
+    {:ok, _} =
+      insert_changes_list(
+        ordered_changes_list,
+        conflict_target: :contract_address_hash,
+        on_conflict: on_conflict,
+        for: Token,
+        returning: true,
+        timeout: timeout,
+        timestamps: timestamps
+      )
+  end
+
+  @spec insert_token_transfers([map()], %{required(:timeout) => timeout(), required(:timestamps) => timestamps()}) ::
+          {:ok, [TokenTransfer.t()]}
+          | {:error, [Changeset.t()]}
+  def insert_token_transfers(changes_list, %{timeout: timeout, timestamps: timestamps})
+      when is_list(changes_list) do
+    # order so that row ShareLocks are grabbed in a consistent order
+    ordered_changes_list = Enum.sort_by(changes_list, &{&1.transaction_hash, &1.log_index})
+
+    {:ok, _} =
+      insert_changes_list(
+        ordered_changes_list,
+        conflict_target: [:transaction_hash, :log_index],
+        on_conflict: :replace_all,
+        for: TokenTransfer,
+        returning: true,
+        timeout: timeout,
+        timestamps: timestamps
+      )
+  end
+
   @spec insert_transactions([map()], %{
           required(:on_conflict) => on_conflict,
           required(:timeout) => timeout,

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

@@ -1,6 +1,20 @@
 defmodule Explorer.Chain.Token do
   @moduledoc """
-  Represents an ERC-20 token.
+  Represents a token.
+
+  ## Token Indexing
+
+  The following types of tokens are indexed:
+
+  * ERC-20
+  * ERC-721
+
+  ## Token Specifications
+
+  * [ERC-20](https://github.com/ethereum/EIPs/blob/master/EIPS/eip-20.md)
+  * [ERC-721](https://github.com/ethereum/EIPs/blob/master/EIPS/eip-721.md)
+  * [ERC-777](https://github.com/ethereum/EIPs/blob/master/EIPS/eip-777.md)
+  * [ERC-1155](https://github.com/ethereum/EIPs/blob/master/EIPS/eip-1155.md)
   """
 
   use Ecto.Schema
@@ -13,6 +27,8 @@ defmodule Explorer.Chain.Token do
   * `:symbol` - Trading symbol of the token
   * `:total_supply` - The total supply of the token
   * `:decimals` - Number of decimal places the token can be subdivided to
+  * `:type` - Type of token
+  * `:calatoged` - Flag for if token information has been cataloged
   * `:contract_address` - The `t:Address.t/0` of the token's contract
   * `:contract_address_hash` - Address hash foreign key
   """
@@ -21,15 +37,20 @@ defmodule Explorer.Chain.Token do
           symbol: String.t(),
           total_supply: Decimal.t(),
           decimals: non_neg_integer(),
+          type: String.t(),
+          cataloged: boolean(),
           contract_address: %Ecto.Association.NotLoaded{} | Address.t(),
           contract_address_hash: Hash.Address.t()
         }
 
+  @primary_key false
   schema "tokens" do
     field(:name, :string)
     field(:symbol, :string)
     field(:total_supply, :decimal)
     field(:decimals, :integer)
+    field(:type, :string)
+    field(:cataloged, :boolean)
 
     belongs_to(
       :contract_address,
@@ -42,11 +63,14 @@ defmodule Explorer.Chain.Token do
     timestamps()
   end
 
+  @required_attrs ~w(contract_address_hash type)a
+  @optional_attrs ~w(cataloged decimals name symbol total_supply)a
+
   @doc false
   def changeset(%Token{} = token, params \\ %{}) do
     token
-    |> cast(params, ~w(name symbol total_supply decimals contract_address_hash)a)
-    |> validate_required(~w(contract_address_hash))
+    |> cast(params, @required_attrs ++ @optional_attrs)
+    |> validate_required(@required_attrs)
     |> foreign_key_constraint(:contract_address)
     |> unique_constraint(:contract_address_hash)
   end

apps/explorer/lib/explorer/chain/token_transfer.ex

@@ -36,6 +36,7 @@ defmodule Explorer.Chain.TokenTransfer do
   * `:to_address_hash` - Address hash foreign key
   * `:token_contract_address` - The `t:Explorer.Chain.Address.t/0` of the token's contract.
   * `:token_contract_address_hash` - Address hash foreign key
+  * `:token_id` - ID of the token (applicable to ERC-721 tokens)
   * `:transaction` - The `t:Explorer.Chain.Transaction.t/0` ledger
   * `:transaction_hash` - Transaction foreign key
   * `:log_index` - Index of the corresponding `t:Explorer.Chain.Log.t/0` in the transaction.
@@ -48,6 +49,7 @@ defmodule Explorer.Chain.TokenTransfer do
           to_address_hash: Hash.Address.t(),
           token_contract_address: %Ecto.Association.NotLoaded{} | Address.t(),
           token_contract_address_hash: Hash.Address.t(),
+          token_id: non_neg_integer() | nil,
           transaction: %Ecto.Association.NotLoaded{} | Transaction.t(),
           transaction: %Ecto.Association.NotLoaded{} | Transaction.t(),
           transaction_hash: Hash.Full.t(),
@@ -59,6 +61,7 @@ defmodule Explorer.Chain.TokenTransfer do
   schema "token_transfers" do
     field(:amount, :decimal)
     field(:log_index, :integer)
+    field(:token_id, :integer)
 
     belongs_to(:from_address, Address, foreign_key: :from_address_hash, references: :hash, type: Hash.Address)
     belongs_to(:to_address, Address, foreign_key: :to_address_hash, references: :hash, type: Hash.Address)
@@ -78,11 +81,14 @@ defmodule Explorer.Chain.TokenTransfer do
     timestamps()
   end
 
+  @required_attrs ~w(log_index from_address_hash to_address_hash token_contract_address_hash transaction_hash)a
+  @optional_attrs ~w(amount token_id)a
+
   @doc false
   def changeset(%TokenTransfer{} = struct, params \\ %{}) do
     struct
-    |> cast(params, ~w(amount log_index from_address_hash to_address_hash token_contract_address_hash transaction_hash))
-    |> validate_required(~w(log_index from_address_hash to_address_hash token_contract_address_hash))
+    |> cast(params, @required_attrs ++ @optional_attrs)
+    |> validate_required(@required_attrs)
     |> foreign_key_constraint(:from_address)
     |> foreign_key_constraint(:to_address)
     |> foreign_key_constraint(:token_contract_address)

apps/explorer/lib/explorer/chain/transaction.ex

@@ -81,7 +81,7 @@ defmodule Explorer.Chain.Transaction do
    * `input`- data sent along with the transaction
    * `internal_transactions` - transactions (value transfers) created while executing contract used for this
      transaction
-   * `internal_transactions_indexed_at` - when `internal_transactions` were fetched by `Explorer.Indexer`.
+   * `internal_transactions_indexed_at` - when `internal_transactions` were fetched by `Indexer`.
    * `logs` - events that occurred while mining the `transaction`.
    * `nonce` - the number of transaction made by the sender prior to this one
    * `r` - the R field of the signature. The (r, s) is the normal output of an ECDSA signature, where r is computed as

apps/explorer/lib/explorer/smart_contract/reader.ex

@@ -7,9 +7,23 @@ defmodule Explorer.SmartContract.Reader do
   """
 
   alias Explorer.Chain
-  alias EthereumJSONRPC.Encoder
   alias Explorer.Chain.Hash
+  alias EthereumJSONRPC.Encoder
+
+  @typedoc """
+  Map of functions to call with the values for the function to be called with.
+  """
+  @type functions :: %{String.t() => [term()]}
 
+  @typedoc """
+  Map of function call to function call results.
+  """
+  @type functions_results :: %{String.t() => {:ok, term()} | {:error, String.t()}}
+
+  @typedoc """
+  Options that can be forwarded when calling the Ethereum JSON RPC.
+  """
+  @type contract_call_options :: [{:json_rpc_named_arguments, EthereumJSONRPC.json_rpc_named_arguments()}]
   @doc """
   Queries the contract functions on the blockchain and returns the call results.
 
@@ -18,28 +32,26 @@ defmodule Explorer.SmartContract.Reader do
   Note that for this example to work the database must be up to date with the
   information available in the blockchain.
 
-  ```
-  $ Explorer.SmartContract.Reader.query_contract(
+      $ Explorer.SmartContract.Reader.query_verified_contract(
         %Explorer.Chain.Hash{
           byte_count: 20,
           bytes: <<0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0>>
         },
         %{"sum" => [20, 22]}
       )
-  # => %{"sum" => [42]}
+      # => %{"sum" => {:ok, 42}}
 
-  $ Explorer.SmartContract.Reader.query_contract(
+      $ Explorer.SmartContract.Reader.query_verified_contract(
         %Explorer.Chain.Hash{
           byte_count: 20,
           bytes: <<0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0>>
         },
         %{"sum" => [1, "abc"]}
       )
-  # => %{"sum" => ["Data overflow encoding int, data `abc` cannot fit in 256 bits"]}
-  ```
+      # => %{"sum" => {:error, "Data overflow encoding int, data `abc` cannot fit in 256 bits"}}
   """
-  @spec query_contract(%Explorer.Chain.Hash{}, %{String.t() => [term()]}) :: map()
-  def query_contract(address_hash, functions) do
+  @spec query_verified_contract(Hash.Address.t(), functions()) :: functions_results()
+  def query_verified_contract(address_hash, functions) do
     json_rpc_named_arguments = Application.get_env(:explorer, :json_rpc_named_arguments)
 
     contract_address = Hash.to_string(address_hash)
@@ -49,7 +61,36 @@ defmodule Explorer.SmartContract.Reader do
       |> Chain.address_hash_to_smart_contract()
       |> Map.get(:abi)
 
-    try do
+    query_contract(contract_address, abi, functions, json_rpc_named_arguments)
+  end
+
+  @doc """
+  Runs contract functions on a given address for an unverified contract with an expected ABI.
+
+  ## Options
+
+  * `:json_rpc_named_arguments` - Options to forward for calling the Ethereum JSON RPC. See
+    `t:EthereumJSONRPC.json_rpc_named_arguments.t/0` for full list of options.
+  """
+  @spec query_unverified_contract(Hash.Address.t(), [map()], functions(), contract_call_options()) ::
+          functions_results()
+  def query_unverified_contract(
+        %Hash{byte_count: unquote(Hash.Address.byte_count())} = address,
+        abi,
+        functions,
+        opts \\ []
+      ) do
+    contract_address = Hash.to_string(address)
+
+    json_rpc_named_arguments =
+      Keyword.get(opts, :json_rpc_named_arguments) || Application.get_env(:explorer, :json_rpc_named_arguments)
+
+    query_contract(contract_address, abi, functions, json_rpc_named_arguments)
+  end
+
+  @spec query_contract(String.t(), term(), functions(), EthereumJSONRPC.json_rpc_named_arguments()) ::
+          functions_results()
+  defp query_contract(contract_address, abi, functions, json_rpc_named_arguments) do
     blockchain_result =
       abi
       |> Encoder.encode_abi(functions)
@@ -61,12 +102,11 @@ defmodule Explorer.SmartContract.Reader do
     error ->
       format_error(functions, error.message)
   end
-  end
 
   defp format_error(functions, message) do
     functions
     |> Enum.map(fn {function_name, _args} ->
-      %{function_name => [message]}
+      %{function_name => {:error, message}}
     end)
     |> List.first()
   end
@@ -114,7 +154,7 @@ defmodule Explorer.SmartContract.Reader do
         }
       ]
   """
-  @spec read_only_functions(%Explorer.Chain.Hash{}) :: [%{}]
+  @spec read_only_functions(Hash.t()) :: [%{}]
   def read_only_functions(contract_address_hash) do
     contract_address_hash
     |> Chain.address_hash_to_smart_contract()
@@ -155,7 +195,7 @@ defmodule Explorer.SmartContract.Reader do
     query_function(contract_address_hash, %{name: name, args: []})
   end
 
-  @spec query_function(%Explorer.Chain.Hash{}, %{name: String.t(), args: [term()]}) :: [%{}]
+  @spec query_function(Hash.t(), %{name: String.t(), args: [term()]}) :: [%{}]
   def query_function(contract_address_hash, %{name: name, args: args}) do
     function =
       contract_address_hash
@@ -169,7 +209,7 @@ defmodule Explorer.SmartContract.Reader do
 
   defp fetch_from_blockchain(contract_address_hash, %{name: name, args: args, outputs: outputs}) do
     contract_address_hash
-    |> query_contract(%{name => normalize_args(args)})
+    |> query_verified_contract(%{name => normalize_args(args)})
     |> link_outputs_and_values(outputs, name)
   end
 
@@ -194,9 +234,9 @@ defmodule Explorer.SmartContract.Reader do
   end
 
   def link_outputs_and_values(blockchain_values, outputs, function_name) do
-    values = Map.get(blockchain_values, function_name, [""])
+    {_, value} = Map.get(blockchain_values, function_name, {:ok, ""})
 
-    for output <- outputs, value <- values do
+    for output <- outputs do
       new_value(output, value)
     end
   end

apps/explorer/mix.exs

@@ -71,6 +71,9 @@ defmodule Explorer.Mixfile do
       {:credo, "0.9.2", only: [:dev, :test], runtime: false},
       {:crontab, "~> 1.1"},
       {:dialyxir, "~> 0.5", only: [:dev, :test], runtime: false},
+      # Casting Ethereum-native types to Elixir-native types
+      {:ecto, "~> 2.2"},
+      # Data factory for testing
       {:ex_machina, "~> 2.1", only: [:test]},
       # Code coverage
       {:excoveralls, "~> 0.8.1", only: [:test]},
@@ -80,7 +83,7 @@ defmodule Explorer.Mixfile do
       {:junit_formatter, ">= 0.0.0", only: [:test], runtime: false},
       {:math, "~> 0.3.0"},
       {:mock, "~> 0.3.0", only: [:test], runtime: false},
-      {:mox, "~> 0.3.2", only: [:test]},
+      {:mox, "~> 0.4", only: [:test]},
       {:postgrex, ">= 0.0.0"},
       {:sobelow, ">= 0.7.0", only: [:dev, :test], runtime: false},
       {:timex, "~> 3.1.24"},

apps/explorer/priv/repo/migrations/20180606135149_create_tokens.exs

@@ -2,11 +2,15 @@ defmodule Explorer.Repo.Migrations.CreateTokens do
   use Ecto.Migration
 
   def change do
-    create table(:tokens) do
-      add(:name, :string)
-      add(:symbol, :string)
-      add(:total_supply, :decimal)
-      add(:decimals, :smallint)
+    create table(:tokens, primary_key: false) do
+      # Name, symbol, total supply, and decimals may not always be available from executing a token contract
+      # Allow for nulls for those fields
+      add(:name, :string, null: true)
+      add(:symbol, :string, null: true)
+      add(:total_supply, :decimal, null: true)
+      add(:decimals, :smallint, null: true)
+      add(:type, :string, null: false)
+      add(:cataloged, :boolean, default: false)
 
       add(
         :contract_address_hash,

apps/explorer/priv/repo/migrations/20180606135150_create_token_transfers.exs

@@ -12,7 +12,10 @@ defmodule Explorer.Repo.Migrations.CreateTokenTransfers do
       add(:log_index, :integer, null: false)
       add(:from_address_hash, references(:addresses, column: :hash, type: :bytea), null: false)
       add(:to_address_hash, references(:addresses, column: :hash, type: :bytea), null: false)
-      add(:amount, :decimal, null: false)
+      # Some token transfers do not have a fungible value like ERC721 transfers
+      add(:amount, :decimal, null: true)
+      # ERC-721 tokens have IDs
+      add(:token_id, :integer, null: true)
       add(:token_contract_address_hash, references(:addresses, column: :hash, type: :bytea), null: false)
 
       timestamps()

apps/explorer/test/explorer/chain/data_test.exs

@@ -1,4 +1,4 @@
-defmodule Explrer.Chain.DataTest do
+defmodule Explorer.Chain.DataTest do
   use ExUnit.Case, async: true
 
   doctest Explorer.Chain.Data

apps/explorer/test/explorer/chain_test.exs

@@ -4,7 +4,18 @@ defmodule Explorer.ChainTest do
   import Explorer.Factory
 
   alias Explorer.{Chain, Factory, PagingOptions, Repo}
-  alias Explorer.Chain.{Address, Block, Hash, InternalTransaction, Log, SmartContract, Transaction, Wei}
+
+  alias Explorer.Chain.{
+    Address,
+    Block,
+    InternalTransaction,
+    Log,
+    Token,
+    Transaction,
+    SmartContract,
+    Wei
+  }
+
   alias Explorer.Chain.Supply.ProofOfAuthority
 
   doctest Explorer.Chain
@@ -1488,123 +1499,22 @@ defmodule Explorer.ChainTest do
     assert [{^current_pid, _}] = Registry.lookup(Registry.ChainEvents, :logs)
   end
 
-  describe "import" do
-    @import_data %{
-      blocks: %{
-        params: [
-          %{
-            difficulty: 340_282_366_920_938_463_463_374_607_431_768_211_454,
-            gas_limit: 6_946_336,
-            gas_used: 50450,
-            hash: "0xf6b4b8c88df3ebd252ec476328334dc026cf66606a84fb769b3d3cbccc8471bd",
-            miner_hash: "0xe8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca",
-            nonce: 0,
-            number: 37,
-            parent_hash: "0xc37bbad7057945d1bf128c1ff009fb1ad632110bf6a000aac025a80f7766b66e",
-            size: 719,
-            timestamp: Timex.parse!("2017-12-15T21:06:30.000000Z", "{ISO:Extended:Z}"),
-            total_difficulty: 12_590_447_576_074_723_148_144_860_474_975_121_280_509
-          }
-        ]
-      },
-      broadcast: true,
-      internal_transactions: %{
-        params: [
-          %{
-            call_type: "call",
-            from_address_hash: "0xe8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca",
-            gas: 4_677_320,
-            gas_used: 27770,
-            index: 0,
-            output: "0x",
-            to_address_hash: "0x8bf38d4764929064f2d4d3a56520a76ab3df415b",
-            trace_address: [],
-            transaction_hash: "0x53bd884872de3e488692881baeec262e7b95234d3965248c39fe992fffd433e5",
-            type: "call",
-            value: 0
-          }
-        ]
-      },
-      logs: %{
-        params: [
-          %{
-            address_hash: "0x8bf38d4764929064f2d4d3a56520a76ab3df415b",
-            data: "0x000000000000000000000000862d67cb0773ee3f8ce7ea89b328ffea861ab3ef",
-            first_topic: "0x600bcf04a13e752d1e3670a5a9f1c21177ca2a93c6f5391d4f1298d098097c22",
-            fourth_topic: nil,
-            index: 0,
-            second_topic: nil,
-            third_topic: nil,
-            transaction_hash: "0x53bd884872de3e488692881baeec262e7b95234d3965248c39fe992fffd433e5",
-            type: "mined"
-          }
-        ]
-      },
-      transactions: %{
-        on_conflict: :replace_all,
-        params: [
-          %{
-            block_hash: "0xf6b4b8c88df3ebd252ec476328334dc026cf66606a84fb769b3d3cbccc8471bd",
-            block_number: 37,
-            cumulative_gas_used: 50450,
-            from_address_hash: "0xe8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca",
-            gas: 4_700_000,
-            gas_price: 100_000_000_000,
-            gas_used: 50450,
-            hash: "0x53bd884872de3e488692881baeec262e7b95234d3965248c39fe992fffd433e5",
-            index: 0,
-            input: "0x10855269000000000000000000000000862d67cb0773ee3f8ce7ea89b328ffea861ab3ef",
-            nonce: 4,
-            public_key:
-              "0xe5d196ad4ceada719d9e592f7166d0c75700f6eab2e3c3de34ba751ea786527cb3f6eb96ad9fdfdb9989ff572df50f1c42ef800af9c5207a38b929aff969b5c9",
-            r: 0xA7F8F45CCE375BB7AF8750416E1B03E0473F93C256DA2285D1134FC97A700E01,
-            s: 0x1F87A076F13824F4BE8963E3DFFD7300DAE64D5F23C9A062AF0C6EAD347C135F,
-            standard_v: 1,
-            status: :ok,
-            to_address_hash: "0x8bf38d4764929064f2d4d3a56520a76ab3df415b",
-            v: 0xBE,
-            value: 0
-          }
-        ]
-      },
-      addresses: %{
-        params: [
-          %{hash: "0x8bf38d4764929064f2d4d3a56520a76ab3df415b"},
-          %{hash: "0xe8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca"}
-        ]
-      }
-    }
-
-    test "publishes addresses with updated fetched_balance data to subscribers on insert" do
-      Chain.subscribe_to_events(:addresses)
-      Chain.import(@import_data)
-      assert_received {:chain_event, :addresses, [%Address{}, %Address{}]}
+  describe "token_from_address_hash/1" do
+    test "with valid hash" do
+      token = insert(:token)
+      assert {:ok, result} = Chain.token_from_address_hash(token.contract_address.hash)
+      assert result.contract_address_hash == token.contract_address_hash
     end
 
-    test "publishes block data to subscribers on insert" do
-      Chain.subscribe_to_events(:blocks)
-      Chain.import(@import_data)
-      assert_received {:chain_event, :blocks, [%Block{}]}
+    test "with hash that doesn't exist" do
+      token = build(:token)
+      assert {:error, :not_found} = Chain.token_from_address_hash(token.contract_address.hash)
     end
-
-    test "publishes log data to subscribers on insert" do
-      Chain.subscribe_to_events(:logs)
-      Chain.import(@import_data)
-      assert_received {:chain_event, :logs, [%Log{}]}
   end
 
-    test "publishes transaction hashes data to subscribers on insert" do
-      Chain.subscribe_to_events(:transactions)
-      Chain.import(@import_data)
-      assert_received {:chain_event, :transactions, [%Hash{}]}
-    end
-
-    test "does not broadcast if broadcast option is false" do
-      non_broadcast_data = Map.put(@import_data, :broadcast, false)
-
-      Chain.subscribe_to_events(:logs)
-      Chain.import(non_broadcast_data)
-      refute_received {:chain_event, :logs, [%Log{}]}
-    end
+  test "stream_uncataloged_token_contract_address_hashes/2 reduces with given reducer and accumulator" do
+    insert(:token, cataloged: true)
+    %Token{contract_address_hash: uncatalog_address} = insert(:token, cataloged: false)
+    assert Chain.stream_uncataloged_token_contract_address_hashes([], &[&1 | &2]) == {:ok, [uncatalog_address]}
   end
 end

apps/explorer/test/explorer/import_test.exs

@@ -2,11 +2,367 @@ defmodule Explorer.Chain.ImportTest do
   use Explorer.DataCase
 
   alias Explorer.Chain
-  alias Explorer.Chain.{Address, Import, Transaction}
+
+  alias Explorer.Chain.{
+    Address,
+    Block,
+    Data,
+    Log,
+    Hash,
+    Import,
+    Token,
+    TokenTransfer,
+    Transaction
+  }
 
   doctest Import
 
   describe "all/1" do
+    @import_data %{
+      blocks: %{
+        params: [
+          %{
+            difficulty: 340_282_366_920_938_463_463_374_607_431_768_211_454,
+            gas_limit: 6_946_336,
+            gas_used: 50450,
+            hash: "0xf6b4b8c88df3ebd252ec476328334dc026cf66606a84fb769b3d3cbccc8471bd",
+            miner_hash: "0xe8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca",
+            nonce: 0,
+            number: 37,
+            parent_hash: "0xc37bbad7057945d1bf128c1ff009fb1ad632110bf6a000aac025a80f7766b66e",
+            size: 719,
+            timestamp: Timex.parse!("2017-12-15T21:06:30.000000Z", "{ISO:Extended:Z}"),
+            total_difficulty: 12_590_447_576_074_723_148_144_860_474_975_121_280_509
+          }
+        ]
+      },
+      broadcast: true,
+      internal_transactions: %{
+        params: [
+          %{
+            call_type: "call",
+            from_address_hash: "0xe8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca",
+            gas: 4_677_320,
+            gas_used: 27770,
+            index: 0,
+            output: "0x",
+            to_address_hash: "0x8bf38d4764929064f2d4d3a56520a76ab3df415b",
+            trace_address: [],
+            transaction_hash: "0x53bd884872de3e488692881baeec262e7b95234d3965248c39fe992fffd433e5",
+            type: "call",
+            value: 0
+          }
+        ]
+      },
+      logs: %{
+        params: [
+          %{
+            address_hash: "0x8bf38d4764929064f2d4d3a56520a76ab3df415b",
+            data: "0x0000000000000000000000000000000000000000000000000de0b6b3a7640000",
+            first_topic: "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",
+            second_topic: "0x000000000000000000000000e8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca",
+            third_topic: "0x000000000000000000000000515c09c5bba1ed566b02a5b0599ec5d5d0aee73d",
+            fourth_topic: nil,
+            index: 0,
+            transaction_hash: "0x53bd884872de3e488692881baeec262e7b95234d3965248c39fe992fffd433e5",
+            type: "mined"
+          }
+        ]
+      },
+      transactions: %{
+        on_conflict: :replace_all,
+        params: [
+          %{
+            block_hash: "0xf6b4b8c88df3ebd252ec476328334dc026cf66606a84fb769b3d3cbccc8471bd",
+            block_number: 37,
+            cumulative_gas_used: 50450,
+            from_address_hash: "0xe8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca",
+            gas: 4_700_000,
+            gas_price: 100_000_000_000,
+            gas_used: 50450,
+            hash: "0x53bd884872de3e488692881baeec262e7b95234d3965248c39fe992fffd433e5",
+            index: 0,
+            input: "0x10855269000000000000000000000000862d67cb0773ee3f8ce7ea89b328ffea861ab3ef",
+            nonce: 4,
+            public_key:
+              "0xe5d196ad4ceada719d9e592f7166d0c75700f6eab2e3c3de34ba751ea786527cb3f6eb96ad9fdfdb9989ff572df50f1c42ef800af9c5207a38b929aff969b5c9",
+            r: 0xA7F8F45CCE375BB7AF8750416E1B03E0473F93C256DA2285D1134FC97A700E01,
+            s: 0x1F87A076F13824F4BE8963E3DFFD7300DAE64D5F23C9A062AF0C6EAD347C135F,
+            standard_v: 1,
+            status: :ok,
+            to_address_hash: "0x8bf38d4764929064f2d4d3a56520a76ab3df415b",
+            v: 0xBE,
+            value: 0
+          }
+        ]
+      },
+      addresses: %{
+        params: [
+          %{hash: "0x8bf38d4764929064f2d4d3a56520a76ab3df415b"},
+          %{hash: "0xe8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca"},
+          %{hash: "0x515c09c5bba1ed566b02a5b0599ec5d5d0aee73d"}
+        ]
+      },
+      tokens: %{
+        on_conflict: :nothing,
+        params: [
+          %{
+            contract_address_hash: "0x8bf38d4764929064f2d4d3a56520a76ab3df415b",
+            type: "ERC-20"
+          }
+        ]
+      },
+      token_transfers: %{
+        params: [
+          %{
+            amount: Decimal.new(1_000_000_000_000_000_000),
+            block_number: 37,
+            log_index: 0,
+            from_address_hash: "0xe8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca",
+            to_address_hash: "0x515c09c5bba1ed566b02a5b0599ec5d5d0aee73d",
+            token_contract_address_hash: "0x8bf38d4764929064f2d4d3a56520a76ab3df415b",
+            transaction_hash: "0x53bd884872de3e488692881baeec262e7b95234d3965248c39fe992fffd433e5"
+          }
+        ]
+      }
+    }
+
+    test "with valid data" do
+      difficulty = Decimal.new(340_282_366_920_938_463_463_374_607_431_768_211_454)
+      total_difficulty = Decimal.new(12_590_447_576_074_723_148_144_860_474_975_121_280_509)
+      token_transfer_amount = Decimal.new(1_000_000_000_000_000_000)
+
+      assert {:ok,
+              %{
+                addresses: [
+                  %Address{
+                    hash: %Hash{
+                      byte_count: 20,
+                      bytes:
+                        <<81, 92, 9, 197, 187, 161, 237, 86, 107, 2, 165, 176, 89, 158, 197, 213, 208, 174, 231, 61>>
+                    },
+                    inserted_at: %{},
+                    updated_at: %{}
+                  },
+                  %Address{
+                    hash: %Hash{
+                      byte_count: 20,
+                      bytes:
+                        <<139, 243, 141, 71, 100, 146, 144, 100, 242, 212, 211, 165, 101, 32, 167, 106, 179, 223, 65,
+                          91>>
+                    },
+                    inserted_at: %{},
+                    updated_at: %{}
+                  },
+                  %Address{
+                    hash: %Hash{
+                      byte_count: 20,
+                      bytes:
+                        <<232, 221, 197, 199, 162, 210, 240, 215, 169, 121, 132, 89, 192, 16, 79, 223, 94, 152, 122,
+                          202>>
+                    },
+                    inserted_at: %{},
+                    updated_at: %{}
+                  }
+                ],
+                blocks: [
+                  %Block{
+                    difficulty: ^difficulty,
+                    gas_limit: 6_946_336,
+                    gas_used: 50450,
+                    hash: %Hash{
+                      byte_count: 32,
+                      bytes:
+                        <<246, 180, 184, 200, 141, 243, 235, 210, 82, 236, 71, 99, 40, 51, 77, 192, 38, 207, 102, 96,
+                          106, 132, 251, 118, 155, 61, 60, 188, 204, 132, 113, 189>>
+                    },
+                    miner_hash: %Hash{
+                      byte_count: 20,
+                      bytes:
+                        <<232, 221, 197, 199, 162, 210, 240, 215, 169, 121, 132, 89, 192, 16, 79, 223, 94, 152, 122,
+                          202>>
+                    },
+                    nonce: %Explorer.Chain.Hash{
+                      byte_count: 8,
+                      bytes: <<0, 0, 0, 0, 0, 0, 0, 0>>
+                    },
+                    number: 37,
+                    parent_hash: %Hash{
+                      byte_count: 32,
+                      bytes:
+                        <<195, 123, 186, 215, 5, 121, 69, 209, 191, 18, 140, 31, 240, 9, 251, 26, 214, 50, 17, 11, 246,
+                          160, 0, 170, 192, 37, 168, 15, 119, 102, 182, 110>>
+                    },
+                    size: 719,
+                    timestamp: %DateTime{
+                      year: 2017,
+                      month: 12,
+                      day: 15,
+                      hour: 21,
+                      minute: 6,
+                      second: 30,
+                      microsecond: {0, 6},
+                      std_offset: 0,
+                      utc_offset: 0,
+                      time_zone: "Etc/UTC",
+                      zone_abbr: "UTC"
+                    },
+                    total_difficulty: ^total_difficulty,
+                    inserted_at: %{},
+                    updated_at: %{}
+                  }
+                ],
+                internal_transactions: [
+                  %{
+                    index: 0,
+                    transaction_hash: %Hash{
+                      byte_count: 32,
+                      bytes:
+                        <<83, 189, 136, 72, 114, 222, 62, 72, 134, 146, 136, 27, 174, 236, 38, 46, 123, 149, 35, 77, 57,
+                          101, 36, 140, 57, 254, 153, 47, 255, 212, 51, 229>>
+                    }
+                  }
+                ],
+                logs: [
+                  %Log{
+                    address_hash: %Hash{
+                      byte_count: 20,
+                      bytes:
+                        <<139, 243, 141, 71, 100, 146, 144, 100, 242, 212, 211, 165, 101, 32, 167, 106, 179, 223, 65,
+                          91>>
+                    },
+                    data: %Data{
+                      bytes:
+                        <<0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 13, 224, 182, 179,
+                          167, 100, 0, 0>>
+                    },
+                    index: 0,
+                    first_topic: "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",
+                    second_topic: "0x000000000000000000000000e8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca",
+                    third_topic: "0x000000000000000000000000515c09c5bba1ed566b02a5b0599ec5d5d0aee73d",
+                    fourth_topic: nil,
+                    transaction_hash: %Hash{
+                      byte_count: 32,
+                      bytes:
+                        <<83, 189, 136, 72, 114, 222, 62, 72, 134, 146, 136, 27, 174, 236, 38, 46, 123, 149, 35, 77, 57,
+                          101, 36, 140, 57, 254, 153, 47, 255, 212, 51, 229>>
+                    },
+                    type: "mined",
+                    inserted_at: %{},
+                    updated_at: %{}
+                  }
+                ],
+                transactions: [
+                  %Hash{
+                    byte_count: 32,
+                    bytes:
+                      <<83, 189, 136, 72, 114, 222, 62, 72, 134, 146, 136, 27, 174, 236, 38, 46, 123, 149, 35, 77, 57,
+                        101, 36, 140, 57, 254, 153, 47, 255, 212, 51, 229>>
+                  }
+                ],
+                tokens: [
+                  %Token{
+                    contract_address_hash: %Hash{
+                      byte_count: 20,
+                      bytes:
+                        <<139, 243, 141, 71, 100, 146, 144, 100, 242, 212, 211, 165, 101, 32, 167, 106, 179, 223, 65,
+                          91>>
+                    },
+                    type: "ERC-20",
+                    inserted_at: %{},
+                    updated_at: %{}
+                  }
+                ],
+                token_transfers: [
+                  %TokenTransfer{
+                    amount: ^token_transfer_amount,
+                    log_index: 0,
+                    from_address_hash: %Hash{
+                      byte_count: 20,
+                      bytes:
+                        <<232, 221, 197, 199, 162, 210, 240, 215, 169, 121, 132, 89, 192, 16, 79, 223, 94, 152, 122,
+                          202>>
+                    },
+                    to_address_hash: %Hash{
+                      byte_count: 20,
+                      bytes:
+                        <<81, 92, 9, 197, 187, 161, 237, 86, 107, 2, 165, 176, 89, 158, 197, 213, 208, 174, 231, 61>>
+                    },
+                    token_contract_address_hash: %Hash{
+                      byte_count: 20,
+                      bytes:
+                        <<139, 243, 141, 71, 100, 146, 144, 100, 242, 212, 211, 165, 101, 32, 167, 106, 179, 223, 65,
+                          91>>
+                    },
+                    transaction_hash: %Hash{
+                      byte_count: 32,
+                      bytes:
+                        <<83, 189, 136, 72, 114, 222, 62, 72, 134, 146, 136, 27, 174, 236, 38, 46, 123, 149, 35, 77, 57,
+                          101, 36, 140, 57, 254, 153, 47, 255, 212, 51, 229>>
+                    },
+                    inserted_at: %{},
+                    updated_at: %{}
+                  }
+                ]
+              }} = Import.all(@import_data)
+    end
+
+    test "with empty map" do
+      assert {:ok, %{}} == Import.all(%{})
+    end
+
+    test "publishes data to subscribers on insert" do
+      Chain.subscribe_to_events(:logs)
+      Import.all(@import_data)
+      assert_received {:chain_event, :logs, [%Log{}]}
+    end
+
+    test "with invalid data" do
+      invalid_transaction =
+        @import_data
+        |> Map.get(:internal_transactions)
+        |> Map.get(:params)
+        |> Enum.at(0)
+        |> Map.delete(:call_type)
+
+      invalid_import_data = put_in(@import_data, [:internal_transactions, :params], [invalid_transaction])
+
+      assert {:error, [changeset]} = Import.all(invalid_import_data)
+      assert changeset_errors(changeset)[:call_type] == ["can't be blank"]
+    end
+
+    test "publishes addresses with updated fetched_balance data to subscribers on insert" do
+      Chain.subscribe_to_events(:addresses)
+      Import.all(@import_data)
+      assert_received {:chain_event, :addresses, [%Address{}, %Address{}, %Address{}]}
+    end
+
+    test "publishes block data to subscribers on insert" do
+      Chain.subscribe_to_events(:blocks)
+      Import.all(@import_data)
+      assert_received {:chain_event, :blocks, [%Block{}]}
+    end
+
+    test "publishes log data to subscribers on insert" do
+      Chain.subscribe_to_events(:logs)
+      Import.all(@import_data)
+      assert_received {:chain_event, :logs, [%Log{}]}
+    end
+
+    test "publishes transaction hashes data to subscribers on insert" do
+      Chain.subscribe_to_events(:transactions)
+      Import.all(@import_data)
+      assert_received {:chain_event, :transactions, [%Hash{}]}
+    end
+
+    test "does not broadcast if broadcast option is false" do
+      non_broadcast_data = Map.merge(@import_data, %{broadcast: false})
+
+      Chain.subscribe_to_events(:logs)
+      Import.all(non_broadcast_data)
+      refute_received {:chain_event, :logs, [%Log{}]}
+    end
+
     test "updates address with contract code" do
       smart_contract_bytecode =
         "0x608060405234801561001057600080fd5b5060df8061001f6000396000f3006080604052600436106049576000357c0100000000000000000000000000000000000000000000000000000000900463ffffffff16806360fe47b114604e5780636d4ce63c146078575b600080fd5b348015605957600080fd5b5060766004803603810190808035906020019092919050505060a0565b005b348015608357600080fd5b50608a60aa565b6040518082815260200191505060405180910390f35b8060008190555050565b600080549050905600a165627a7a7230582040d82a7379b1ee1632ad4d8a239954fd940277b25628ead95259a85c5eddb2120029"

apps/explorer/test/explorer/smart_contract/reader_test.exs

@@ -10,7 +10,7 @@ defmodule Explorer.SmartContract.ReaderTest do
 
   setup :verify_on_exit!
 
-  describe "query_contract/2" do
+  describe "query_verified_contract/2" do
     test "correctly returns the results of the smart contract functions" do
       hash =
         :smart_contract
@@ -19,7 +19,7 @@ defmodule Explorer.SmartContract.ReaderTest do
 
       blockchain_get_function_mock()
 
-      assert Reader.query_contract(hash, %{"get" => []}) == %{"get" => [0]}
+      assert Reader.query_verified_contract(hash, %{"get" => []}) == %{"get" => {:ok, 0}}
     end
 
     test "won't raise error when there is a problem with the params to consult the blockchain" do
@@ -46,9 +46,54 @@ defmodule Explorer.SmartContract.ReaderTest do
 
       wrong_args = %{"sum" => [1, 1, 1, "abc"]}
 
-      assert %{"sum" => ["Data overflow encoding int, data `abc` cannot fit in 256 bits"]} =
-               Reader.query_contract(smart_contract.address_hash, wrong_args)
+      assert %{"sum" => {:error, "Data overflow encoding int, data `abc` cannot fit in 256 bits"}} =
+               Reader.query_verified_contract(smart_contract.address_hash, wrong_args)
     end
+
+    test "handles errors returned from RPC requests" do
+      %{address_hash: address_hash} = insert(:smart_contract)
+
+      expect(
+        EthereumJSONRPC.Mox,
+        :json_rpc,
+        fn [%{id: id, method: _, params: [%{data: _, to: _}]}], _options ->
+          {:ok, [%{id: id, jsonrpc: "2.0", error: %{code: "12345", message: "Error message"}}]}
+        end
+      )
+
+      assert %{"get" => {:error, "(12345) Error message"}} =
+               Reader.query_verified_contract(address_hash, %{"get" => []})
+    end
+  end
+
+  test "query_unverified_contract/3" do
+    address = insert(:address)
+
+    abi = [
+      %{
+        "constant" => true,
+        "inputs" => [],
+        "name" => "decimals",
+        "outputs" => [
+          %{
+            "name" => "",
+            "type" => "uint8"
+          }
+        ],
+        "payable" => false,
+        "type" => "function"
+      }
+    ]
+
+    expect(
+      EthereumJSONRPC.Mox,
+      :json_rpc,
+      fn [%{id: id, method: _, params: [%{data: _, to: _}]}], _options ->
+        {:ok, [%{id: id, jsonrpc: "2.0", result: "0x0000000000000000000000000000000000000000000000000000000000000012"}]}
+      end
+    )
+
+    assert Reader.query_unverified_contract(address.hash, abi, %{"decimals" => []}) == %{"decimals" => {:ok, 18}}
   end
 
   describe "setup_call_payload/2" do
@@ -153,9 +198,8 @@ defmodule Explorer.SmartContract.ReaderTest do
   describe "link_outputs_and_values/2" do
     test "links the ABI outputs with the values retrieved from the blockchain" do
       blockchain_values = %{
-        "getOwner" => [
-          <<105, 55, 203, 37, 235, 84, 188, 1, 59, 156, 19, 196, 122, 179, 142, 182, 62, 221, 20, 147>>
-        ]
+        "getOwner" =>
+          {:ok, <<105, 55, 203, 37, 235, 84, 188, 1, 59, 156, 19, 196, 122, 179, 142, 182, 62, 221, 20, 147>>}
       }
 
       outputs = [%{"name" => "", "type" => "address"}]
@@ -168,9 +212,8 @@ defmodule Explorer.SmartContract.ReaderTest do
 
     test "correctly shows returns of 'bytes' type" do
       blockchain_values = %{
-        "get" => [
-          <<0, 10, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0>>
-        ]
+        "get" =>
+          {:ok, <<0, 10, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0>>}
       }
 
       outputs = [%{"name" => "", "type" => "bytes32"}]
@@ -188,8 +231,8 @@ defmodule Explorer.SmartContract.ReaderTest do
   end
 
   defp blockchain_get_function_mock() do
-    EthereumJSONRPC.Mox
-    |> expect(
+    expect(
+      EthereumJSONRPC.Mox,
       :json_rpc,
       fn [%{id: id, method: _, params: [%{data: _, to: _}]}], _options ->
         {:ok, [%{id: id, jsonrpc: "2.0", result: "0x0000000000000000000000000000000000000000000000000000000000000000"}]}

apps/explorer/test/support/factory.ex

@@ -261,7 +261,9 @@ defmodule Explorer.Factory do
       symbol: "IT",
       total_supply: 1_000_000_000,
       decimals: 18,
-      contract_address: build(:address)
+      contract_address: build(:address),
+      type: "ERC-20",
+      cataloged: true
     }
   end
 

apps/explorer_web/lib/explorer_web/endpoint.ex

@@ -52,12 +52,6 @@ defmodule ExplorerWeb.Endpoint do
 
   plug(ExplorerWeb.Router)
 
-  @doc """
-  Callback invoked for dynamically configuring the endpoint.
-
-  It receives the endpoint configuration and checks if
-  configuration should be loaded from the system environment.
-  """
   def init(_key, config) do
     if config[:load_from_system_env] do
       port = System.get_env("PORT") || raise "expected the PORT environment variable to be set"

apps/explorer_web/lib/explorer_web/templates/address/overview.html.eex

@@ -44,7 +44,7 @@
                         @address.contracts_creation_internal_transaction.transaction_hash
                       ),
                       "data-test": "transaction_hash_link",
-                      "class": "tile-title"
+                      class: "tile-title"
                     ) %>
               </span>
             <% end %>

apps/explorer_web/lib/explorer_web/templates/transaction/_link.html.eex

@@ -1,4 +1,4 @@
 <%= link(@transaction_hash,
     to: transaction_path(ExplorerWeb.Endpoint, :show, @locale, @transaction_hash),
     "data-test": "transaction_hash_link",
-    "class": "text-truncate") %>
+    class: "text-truncate") %>

apps/indexer/lib/indexer/address_extraction.ex

@@ -92,6 +92,20 @@ defmodule Indexer.AddressExtraction do
         %{from: :block_number, to: :fetched_balance_block_number},
         %{from: :address_hash, to: :hash}
       ]
+    ],
+    token_transfers: [
+      [
+        %{from: :block_number, to: :fetched_balance_block_number},
+        %{from: :from_address_hash, to: :hash}
+      ],
+      [
+        %{from: :block_number, to: :fetched_balance_block_number},
+        %{from: :to_address_hash, to: :hash}
+      ],
+      [
+        %{from: :block_number, to: :fetched_balance_block_number},
+        %{from: :token_contract_address_hash, to: :hash}
+      ]
     ]
   }
 
@@ -340,6 +354,14 @@ defmodule Indexer.AddressExtraction do
               required(:address_hash) => String.t(),
               required(:block_number) => non_neg_integer()
             }
+          ],
+          optional(:token_transfers) => [
+            %{
+              required(:from_address_hash) => String.t(),
+              required(:to_address_hash) => String.t(),
+              required(:token_contract_address_hash) => String.t(),
+              required(:block_number) => non_neg_integer()
+            }
           ]
         }) :: [params]
   def extract_addresses(fetched_data, options \\ []) when is_map(fetched_data) and is_list(options) do

apps/indexer/lib/indexer/application.ex

@@ -5,7 +5,13 @@ defmodule Indexer.Application do
 
   use Application
 
-  alias Indexer.{BalanceFetcher, BlockFetcher, InternalTransactionFetcher, PendingTransactionFetcher}
+  alias Indexer.{
+    BalanceFetcher,
+    BlockFetcher,
+    InternalTransactionFetcher,
+    PendingTransactionFetcher,
+    TokenFetcher
+  }
 
   @impl Application
   def start(_type, _args) do
@@ -25,6 +31,7 @@ defmodule Indexer.Application do
       {PendingTransactionFetcher, name: PendingTransactionFetcher, json_rpc_named_arguments: json_rpc_named_arguments},
       {InternalTransactionFetcher,
        name: InternalTransactionFetcher, json_rpc_named_arguments: json_rpc_named_arguments},
+      {TokenFetcher, name: TokenFetcher, json_rpc_named_arguments: json_rpc_named_arguments},
       {BlockFetcher.Supervisor, [block_fetcher_supervisor_named_arguments, [name: BlockFetcher.Supervisor]]}
     ]
 

apps/indexer/lib/indexer/block_fetcher.ex

@@ -8,7 +8,13 @@ defmodule Indexer.BlockFetcher do
   import Indexer, only: [debug: 1]
 
   alias Explorer.Chain.{Block, Import}
-  alias Indexer.{AddressExtraction, Sequence}
+
+  alias Indexer.{
+    AddressExtraction,
+    Sequence,
+    TokenTransfers
+  }
+
   alias Indexer.BlockFetcher.Receipts
 
   # dialyzer thinks that Logger.debug functions always have no_local_return
@@ -41,6 +47,8 @@ defmodule Indexer.BlockFetcher do
                 broadcast: boolean,
                 logs: Import.logs_options(),
                 receipts: Import.receipts_options(),
+                token_transfers: Import.token_transfers_options(),
+                tokens: Import.tokens_options(),
                 transactions: Import.transactions_options()
               }
             ) :: Import.all_result()
@@ -185,11 +193,13 @@ defmodule Indexer.BlockFetcher do
          cap_seq(seq, next, range),
          {:receipts, {:ok, receipt_params}} <- {:receipts, Receipts.fetch(state, transactions_without_receipts)},
          %{logs: logs, receipts: receipts} = receipt_params,
-         transactions_with_receipts = Receipts.put(transactions_without_receipts, receipts) do
+         transactions_with_receipts = Receipts.put(transactions_without_receipts, receipts),
+         %{token_transfers: token_transfers, tokens: tokens} = TokenTransfers.from_log_params(logs) do
       addresses =
         AddressExtraction.extract_addresses(%{
           blocks: blocks,
           logs: logs,
+          token_transfers: token_transfers,
           transactions: transactions_with_receipts
         })
 
@@ -201,6 +211,8 @@ defmodule Indexer.BlockFetcher do
           blocks: %{params: blocks},
           logs: %{params: logs},
           receipts: %{params: receipts},
+          token_transfers: %{params: token_transfers},
+          tokens: %{on_conflict: :nothing, params: tokens},
           transactions: %{params: transactions_with_receipts, on_conflict: :replace_all}
         }
       )

apps/indexer/lib/indexer/block_fetcher/catchup.ex

@@ -9,7 +9,15 @@ defmodule Indexer.BlockFetcher.Catchup do
   import Indexer.BlockFetcher, only: [stream_import: 1]
 
   alias Explorer.Chain
-  alias Indexer.{BalanceFetcher, BlockFetcher, BoundInterval, InternalTransactionFetcher, Sequence}
+
+  alias Indexer.{
+    BalanceFetcher,
+    BlockFetcher,
+    BoundInterval,
+    InternalTransactionFetcher,
+    Sequence,
+    TokenFetcher
+  }
 
   @behaviour BlockFetcher
 
@@ -148,10 +156,13 @@ defmodule Indexer.BlockFetcher.Catchup do
     put_in(supervisor_state.catchup.task, nil)
   end
 
-  defp async_import_remaining_block_data(%{transactions: transaction_hashes, addresses: address_hashes}, %{
+  defp async_import_remaining_block_data(
+         %{transactions: transaction_hashes, addresses: address_hashes, tokens: tokens},
+         %{
            address_hash_to_fetched_balance_block_number: address_hash_to_block_number,
            transaction_hash_to_block_number: transaction_hash_to_block_number
-       }) do
+         }
+       ) do
     address_hashes
     |> Enum.map(fn address_hash ->
       block_number = Map.fetch!(address_hash_to_block_number, to_string(address_hash))
@@ -165,5 +176,9 @@ defmodule Indexer.BlockFetcher.Catchup do
       %{block_number: block_number, hash: transaction_hash}
     end)
     |> InternalTransactionFetcher.async_fetch(10_000)
+
+    tokens
+    |> Enum.map(& &1.contract_address_hash)
+    |> TokenFetcher.async_fetch()
   end
 end

apps/indexer/lib/indexer/block_fetcher/realtime.ex

@@ -9,7 +9,13 @@ defmodule Indexer.BlockFetcher.Realtime do
   import Indexer.BlockFetcher, only: [stream_import: 1]
 
   alias Explorer.Chain
-  alias Indexer.{AddressExtraction, BlockFetcher, Sequence}
+
+  alias Indexer.{
+    AddressExtraction,
+    BlockFetcher,
+    Sequence,
+    TokenFetcher
+  }
 
   @behaviour BlockFetcher
 
@@ -70,13 +76,16 @@ defmodule Indexer.BlockFetcher.Realtime do
            balances(block_fetcher, %{
              address_hash_to_block_number: address_hash_to_block_number,
              address_params: internal_transactions_addresses_params
-           }) do
+           }),
+         chain_import_options =
            options
            |> Map.drop(@import_options)
            |> put_in([:addresses, :params], balances_addresses_params)
            |> put_in([Access.key(:balances, %{}), :params], balances_params)
-      |> put_in([Access.key(:internal_transactions, %{}), :params], internal_transactions_params)
-      |> Chain.import()
+           |> put_in([Access.key(:internal_transactions, %{}), :params], internal_transactions_params),
+         {:ok, results} = ok <- Chain.import(chain_import_options) do
+      async_import_remaining_block_data(results)
+      ok
     end
   end
 
@@ -184,4 +193,10 @@ defmodule Indexer.BlockFetcher.Realtime do
 
     put_in(supervisor_state.realtime.task_by_ref, running_task_by_ref)
   end
+
+  defp async_import_remaining_block_data(%{tokens: tokens}) do
+    tokens
+    |> Enum.map(& &1.contract_address_hash)
+    |> TokenFetcher.async_fetch()
+  end
 end

apps/indexer/lib/indexer/sequence.ex

@@ -212,7 +212,7 @@ defmodule Indexer.Sequence do
     {:ok, reducer.(range, initial)}
   end
 
-  defp reduce_chunked_range(first..last = range, _, step, initial, reducer) do
+  defp reduce_chunked_range(first..last = _range, _, step, initial, reducer) do
     {sign, comparator} =
       if step > 0 do
         {1, &Kernel.>=/2}

apps/indexer/lib/indexer/token_fetcher.ex

@@ -0,0 +1,162 @@
+defmodule Indexer.TokenFetcher do
+  @moduledoc """
+  Fetches information about a token.
+  """
+
+  alias Explorer.Chain
+  alias Explorer.Chain.Token
+  alias Explorer.Chain.Hash.Address
+  alias Explorer.SmartContract.Reader
+  alias Indexer.BufferedTask
+
+  @behaviour BufferedTask
+
+  @defaults [
+    flush_interval: 300,
+    max_batch_size: 1,
+    max_concurrency: 10,
+    init_chunk_size: 1,
+    task_supervisor: Indexer.TaskSupervisor
+  ]
+
+  @contract_abi [
+    %{
+      "constant" => true,
+      "inputs" => [],
+      "name" => "name",
+      "outputs" => [
+        %{
+          "name" => "",
+          "type" => "string"
+        }
+      ],
+      "payable" => false,
+      "type" => "function"
+    },
+    %{
+      "constant" => true,
+      "inputs" => [],
+      "name" => "decimals",
+      "outputs" => [
+        %{
+          "name" => "",
+          "type" => "uint8"
+        }
+      ],
+      "payable" => false,
+      "type" => "function"
+    },
+    %{
+      "constant" => true,
+      "inputs" => [],
+      "name" => "totalSupply",
+      "outputs" => [
+        %{
+          "name" => "",
+          "type" => "uint256"
+        }
+      ],
+      "payable" => false,
+      "type" => "function"
+    },
+    %{
+      "constant" => true,
+      "inputs" => [],
+      "name" => "symbol",
+      "outputs" => [
+        %{
+          "name" => "",
+          "type" => "string"
+        }
+      ],
+      "payable" => false,
+      "type" => "function"
+    }
+  ]
+
+  @doc false
+  def child_spec(provided_opts) do
+    {state, mergable_opts} = Keyword.pop(provided_opts, :json_rpc_named_arguments)
+
+    unless state do
+      raise ArgumentError,
+            ":json_rpc_named_arguments must be provided to `#{__MODULE__}.child_spec " <>
+              "to allow for json_rpc calls when running."
+    end
+
+    opts =
+      @defaults
+      |> Keyword.merge(mergable_opts)
+      |> Keyword.put(:state, state)
+
+    Supervisor.child_spec({BufferedTask, {__MODULE__, opts}}, id: __MODULE__)
+  end
+
+  @impl BufferedTask
+  def init(initial_acc, reducer, _) do
+    {:ok, acc} =
+      Chain.stream_uncataloged_token_contract_address_hashes(initial_acc, fn address, acc ->
+        reducer.(address, acc)
+      end)
+
+    acc
+  end
+
+  @impl BufferedTask
+  def run([token_contract_address], _, json_rpc_named_arguments) do
+    case Chain.token_from_address_hash(token_contract_address) do
+      {:ok, %Token{cataloged: false} = token} ->
+        catalog_token(token, json_rpc_named_arguments)
+
+      {:ok, _} ->
+        :ok
+    end
+  end
+
+  @doc """
+  Fetches token data asynchronously given a list of `t:Explorer.Chain.Token.t/0`s.
+  """
+  @spec async_fetch([Address.t()]) :: :ok
+  def async_fetch(token_contract_addresses) do
+    BufferedTask.buffer(__MODULE__, token_contract_addresses)
+  end
+
+  defp catalog_token(%Token{contract_address_hash: contract_address_hash} = token, json_rpc_named_arguments) do
+    contract_functions = %{
+      "totalSupply" => [],
+      "decimals" => [],
+      "name" => [],
+      "symbol" => []
+    }
+
+    token_contract_results =
+      Reader.query_unverified_contract(
+        contract_address_hash,
+        @contract_abi,
+        contract_functions,
+        json_rpc_named_arguments: json_rpc_named_arguments
+      )
+
+    token_params = format_token_params(token, token_contract_results)
+
+    {:ok, %{tokens: [_]}} = Chain.import(%{tokens: %{params: [token_params], on_conflict: :replace_all}})
+    :ok
+  end
+
+  def format_token_params(token, token_contract_data) do
+    token_contract_data =
+      for {function_name, {:ok, function_data}} <- token_contract_data, into: %{} do
+        {atomized_key(function_name), function_data}
+      end
+
+    token
+    |> Map.from_struct()
+    |> Map.put(:cataloged, true)
+    |> Map.merge(token_contract_data)
+  end
+
+  defp atomized_key("decimals"), do: :decimals
+  defp atomized_key("name"), do: :name
+  defp atomized_key("symbol"), do: :symbol
+  defp atomized_key("totalSupply"), do: :total_supply
+end

apps/indexer/lib/indexer/token_transfers.ex

@@ -0,0 +1,85 @@
+defmodule Indexer.TokenTransfers do
+  @moduledoc """
+  Helper functions for transforming data for ERC-20 and ERC-721 token transfers.
+  """
+
+  alias ABI.TypeDecoder
+  alias Explorer.Chain.TokenTransfer
+
+  @doc """
+  Returns a list of token transfers given a list of logs.
+  """
+  def from_log_params(logs) do
+    initial_acc = %{tokens: [], token_transfers: []}
+
+    logs
+    |> Enum.filter(&(&1.first_topic == unquote(TokenTransfer.constant())))
+    |> Enum.reduce(initial_acc, &do_from_log_params/2)
+  end
+
+  defp do_from_log_params(log, %{tokens: tokens, token_transfers: token_transfers}) do
+    {token, token_transfer} = parse_params(log)
+
+    %{
+      tokens: [token | tokens],
+      token_transfers: [token_transfer | token_transfers]
+    }
+  end
+
+  # ERC-20 token transfer
+  defp parse_params(%{fourth_topic: nil} = log) do
+    token_transfer = %{
+      amount: Decimal.new(convert_to_integer(log.data)),
+      block_number: log.block_number,
+      log_index: log.index,
+      from_address_hash: truncate_address_hash(log.second_topic),
+      to_address_hash: truncate_address_hash(log.third_topic),
+      token_contract_address_hash: log.address_hash,
+      transaction_hash: log.transaction_hash
+    }
+
+    token = %{
+      contract_address_hash: log.address_hash,
+      type: "ERC-20"
+    }
+
+    {token, token_transfer}
+  end
+
+  # ERC-721 token transfer
+  defp parse_params(%{fourth_topic: fourth_topic} = log) when not is_nil(fourth_topic) do
+    token_transfer = %{
+      block_number: log.block_number,
+      log_index: log.index,
+      from_address_hash: truncate_address_hash(log.second_topic),
+      to_address_hash: truncate_address_hash(log.third_topic),
+      token_contract_address_hash: log.address_hash,
+      token_id: convert_to_integer(fourth_topic),
+      transaction_hash: log.transaction_hash
+    }
+
+    token = %{
+      contract_address_hash: log.address_hash,
+      type: "ERC-721"
+    }
+
+    {token, token_transfer}
+  end
+
+  defp truncate_address_hash(nil), do: "0x0000000000000000000000000000000000000000"
+
+  defp truncate_address_hash("0x000000000000000000000000" <> truncated_hash) do
+    "0x#{truncated_hash}"
+  end
+
+  defp convert_to_integer("0x"), do: 0
+
+  defp convert_to_integer("0x" <> encoded_integer) do
+    [value] =
+      encoded_integer
+      |> Base.decode16!(case: :mixed)
+      |> TypeDecoder.decode_raw([{:uint, 256}])
+
+    value
+  end
+end

apps/indexer/mix.exs

@@ -41,7 +41,7 @@ defmodule Indexer.MixProject do
       # Importing to database
       {:explorer, in_umbrella: true},
       # Mocking `EthereumJSONRPC.Transport`, so we avoid hitting real chains for local testing
-      {:mox, "~> 0.3.2", only: [:test]}
+      {:mox, "~> 0.4", only: [:test]}
     ]
   end
 

apps/indexer/test/indexer/address_extraction_test.exs

@@ -100,11 +100,19 @@ defmodule Indexer.AddressExtractionTest do
 
       log = %{address_hash: gen_hash(), block_number: 4}
 
+      token_transfer = %{
+        block_number: 5,
+        from_address_hash: gen_hash(),
+        to_address_hash: gen_hash(),
+        token_contract_address_hash: gen_hash()
+      }
+
       blockchain_data = %{
         blocks: [block],
         internal_transactions: [internal_transaction],
         transactions: [transaction],
-        logs: [log]
+        logs: [log],
+        token_transfers: [token_transfer]
       }
 
       assert AddressExtraction.extract_addresses(blockchain_data) == [
@@ -124,7 +132,19 @@ defmodule Indexer.AddressExtractionTest do
                },
                %{hash: transaction.from_address_hash, fetched_balance_block_number: transaction.block_number},
                %{hash: transaction.to_address_hash, fetched_balance_block_number: transaction.block_number},
-               %{hash: log.address_hash, fetched_balance_block_number: log.block_number}
+               %{hash: log.address_hash, fetched_balance_block_number: log.block_number},
+               %{
+                 hash: token_transfer.from_address_hash,
+                 fetched_balance_block_number: token_transfer.block_number
+               },
+               %{
+                 hash: token_transfer.to_address_hash,
+                 fetched_balance_block_number: token_transfer.block_number
+               },
+               %{
+                 hash: token_transfer.token_contract_address_hash,
+                 fetched_balance_block_number: token_transfer.block_number
+               }
              ]
     end
 

apps/indexer/test/indexer/block_fetcher/realtime_test.exs

@@ -7,6 +7,7 @@ defmodule Indexer.BlockFetcher.RealtimeTest do
   alias Explorer.Chain.{Address, Block}
   alias Indexer.{BlockFetcher, Sequence}
   alias Indexer.BlockFetcher.Realtime
+  alias Indexer.TokenFetcherCase
 
   @moduletag capture_log: true
 
@@ -42,6 +43,9 @@ defmodule Indexer.BlockFetcher.RealtimeTest do
       Sequence.cap(sequence)
       full_block_fetcher = %BlockFetcher{block_fetcher | sequence: sequence}
 
+      start_supervised!({Task.Supervisor, name: Indexer.TaskSupervisor})
+      TokenFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
+
       if json_rpc_named_arguments[:transport] == EthereumJSONRPC.Mox do
         EthereumJSONRPC.Mox
         |> expect(:json_rpc, fn [

apps/indexer/test/indexer/block_fetcher_test.exs

@@ -16,7 +16,8 @@ defmodule Indexer.BlockFetcherTest do
     BufferedTask,
     InternalTransactionFetcher,
     InternalTransactionFetcherCase,
-    Sequence
+    Sequence,
+    TokenFetcherCase
   }
 
   @moduletag capture_log: true
@@ -50,6 +51,7 @@ defmodule Indexer.BlockFetcherTest do
       start_supervised!({Task.Supervisor, name: Indexer.TaskSupervisor})
       AddressBalanceFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
       InternalTransactionFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
+      TokenFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
 
       %{
         block_fetcher:

apps/indexer/test/indexer/supervisor_test.exs

@@ -7,7 +7,15 @@ defmodule Indexer.BlockFetcher.SupervisorTest do
   import EthereumJSONRPC, only: [integer_to_quantity: 1]
 
   alias Explorer.Chain.Block
-  alias Indexer.{AddressBalanceFetcherCase, BlockFetcher, BoundInterval, InternalTransactionFetcherCase}
+
+  alias Indexer.{
+    AddressBalanceFetcherCase,
+    BlockFetcher,
+    BoundInterval,
+    InternalTransactionFetcherCase,
+    TokenFetcherCase
+  }
+
   alias Indexer.BlockFetcher.Catchup
 
   @moduletag capture_log: true
@@ -195,6 +203,7 @@ defmodule Indexer.BlockFetcher.SupervisorTest do
       start_supervised!({Task.Supervisor, name: Indexer.TaskSupervisor})
       AddressBalanceFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
       InternalTransactionFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
+      TokenFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
       start_supervised!({BlockFetcher.Supervisor, [[json_rpc_named_arguments: json_rpc_named_arguments], []]})
 
       first_catchup_block_number = latest_block_number - 1
@@ -238,7 +247,7 @@ defmodule Indexer.BlockFetcher.SupervisorTest do
       start_supervised!({Task.Supervisor, name: Indexer.TaskSupervisor})
       AddressBalanceFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
       InternalTransactionFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
-
+      TokenFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
       # from `setup :state`
       assert_received :catchup_index
 
@@ -307,7 +316,7 @@ defmodule Indexer.BlockFetcher.SupervisorTest do
       start_supervised!({Task.Supervisor, name: Indexer.TaskSupervisor})
       AddressBalanceFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
       InternalTransactionFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
-
+      TokenFetcherCase.start_supervised!(json_rpc_named_arguments: json_rpc_named_arguments)
       # from `setup :state`
       assert_received :catchup_index
 
@@ -317,7 +326,7 @@ defmodule Indexer.BlockFetcher.SupervisorTest do
 
       # 2 blocks are missing, but latest is assumed to be handled by realtime_index, so only 1 is missing for
       # catchup_index
-      assert_receive {^ref, %{first_block_number: 0, missing_block_count: 1}} = message
+      assert_receive {^ref, %{first_block_number: 0, missing_block_count: 1}} = message, 200
 
       # DOWN is not flushed
       assert {:messages, [{:DOWN, ^ref, :process, ^pid, :normal}]} = Process.info(self(), :messages)

apps/indexer/test/indexer/token_fetcher_test.exs

@@ -0,0 +1,78 @@
+defmodule Indexer.TokenFetcherTest do
+  use EthereumJSONRPC.Case
+  use Explorer.DataCase
+
+  import Mox
+
+  alias Explorer.Chain
+  alias Explorer.Chain.Token
+  alias Indexer.TokenFetcher
+
+  setup :verify_on_exit!
+
+  describe "init/3" do
+    test "returns uncataloged tokens", %{json_rpc_named_arguments: json_rpc_named_arguments} do
+      insert(:token, cataloged: true)
+      %Token{contract_address_hash: uncatalog_address} = insert(:token, cataloged: false)
+
+      assert TokenFetcher.init([], &[&1 | &2], json_rpc_named_arguments) == [uncatalog_address]
+    end
+  end
+
+  describe "run/3" do
+    test "skips tokens that have already been cataloged", %{json_rpc_named_arguments: json_rpc_named_arguments} do
+      expect(EthereumJSONRPC.Mox, :json_rpc, 0, fn _, _ -> :ok end)
+      %Token{contract_address_hash: contract_address_hash} = insert(:token, cataloged: true)
+      assert TokenFetcher.run([contract_address_hash], 0, json_rpc_named_arguments) == :ok
+    end
+
+    test "catalogs tokens that haven't been cataloged", %{json_rpc_named_arguments: json_rpc_named_arguments} do
+      token = insert(:token, name: nil, symbol: nil, total_supply: nil, decimals: nil, cataloged: false)
+      contract_address_hash = token.contract_address_hash
+
+      if json_rpc_named_arguments[:transport] == EthereumJSONRPC.Mox do
+        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
+        )
+
+        assert TokenFetcher.run([contract_address_hash], 0, json_rpc_named_arguments) == :ok
+
+        expected_supply = Decimal.new(1_000_000_000_000_000_000)
+
+        assert {:ok,
+                %Token{
+                  name: "Bancor",
+                  symbol: "BNT",
+                  total_supply: ^expected_supply,
+                  decimals: 18,
+                  cataloged: true
+                }} = Chain.token_from_address_hash(contract_address_hash)
+      end
+    end
+  end
+end

apps/indexer/test/indexer/token_transfers_test.exs

@@ -0,0 +1,88 @@
+defmodule Indexer.TokenTransfersTest do
+  use ExUnit.Case
+
+  alias Indexer.TokenTransfers
+
+  describe "from_log_params/2" do
+    test "from_log_params/2 parses logs for tokens and token transfers" do
+      [log_1, _log_2, log_3] =
+        logs = [
+          %{
+            address_hash: "0xf2eec76e45b328df99a34fa696320a262cb92154",
+            block_number: 3_530_917,
+            data: "0x000000000000000000000000000000000000000000000000ebec21ee1da40000",
+            first_topic: "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",
+            fourth_topic: nil,
+            index: 8,
+            second_topic: "0x000000000000000000000000556813d9cc20acfe8388af029a679d34a63388db",
+            third_topic: "0x00000000000000000000000092148dd870fa1b7c4700f2bd7f44238821c26f73",
+            transaction_hash: "0x43dfd761974e8c3351d285ab65bee311454eb45b149a015fe7804a33252f19e5",
+            type: "mined"
+          },
+          %{
+            address_hash: "0x6ea5ec9cb832e60b6b1654f5826e9be638f276a5",
+            block_number: 3_586_935,
+            data: "0x",
+            first_topic: "0x55e10366a5f552746106978b694d7ef3bbddec06bd5f9b9d15ad46f475c653ef",
+            fourth_topic: "0xc89efdaa54c0f20c7adf612882df0950f5a951637e0307cdcb4c672f298b8bc6",
+            index: 0,
+            second_topic: "0x00000000000000000000000063b0595bb7a0b7edd0549c9557a0c8aee6da667b",
+            third_topic: "0x000000000000000000000000f3089e15d0c23c181d7f98b0878b560bfe193a1d",
+            transaction_hash: "0x8425a9b81a9bd1c64861110c1a453b84719cb0361d6fa0db68abf7611b9a890e",
+            type: "mined"
+          },
+          %{
+            address_hash: "0x91932e8c6776fb2b04abb71874a7988747728bb2",
+            block_number: 3_664_064,
+            data: "0x000000000000000000000000000000000000000000000000ebec21ee1da40000",
+            first_topic: "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef",
+            fourth_topic: "0x00000000000000000000000000000000000000000000000000000000000000b7",
+            index: 1,
+            second_topic: "0x0000000000000000000000009851ba177554eb07271ac230a137551e6dd0aa84",
+            third_topic: "0x000000000000000000000000dccb72afee70e60b0c1226288fe86c01b953e8ac",
+            transaction_hash: "0x4011d9a930a3da620321589a54dc0ca3b88216b4886c7a7c3aaad1fb17702d35",
+            type: "mined"
+          }
+        ]
+
+      expected = %{
+        tokens: [
+          %{
+            contract_address_hash: log_3.address_hash,
+            type: "ERC-721"
+          },
+          %{
+            contract_address_hash: log_1.address_hash,
+            type: "ERC-20"
+          }
+        ],
+        token_transfers: [
+          %{
+            block_number: log_3.block_number,
+            log_index: log_3.index,
+            from_address_hash: truncated_hash(log_3.second_topic),
+            to_address_hash: truncated_hash(log_3.third_topic),
+            token_contract_address_hash: log_3.address_hash,
+            token_id: 183,
+            transaction_hash: log_3.transaction_hash
+          },
+          %{
+            amount: Decimal.new(17_000_000_000_000_000_000),
+            block_number: log_1.block_number,
+            log_index: log_1.index,
+            from_address_hash: truncated_hash(log_1.second_topic),
+            to_address_hash: truncated_hash(log_1.third_topic),
+            token_contract_address_hash: log_1.address_hash,
+            transaction_hash: log_1.transaction_hash
+          }
+        ]
+      }
+
+      assert TokenTransfers.from_log_params(logs) == expected
+    end
+  end
+
+  defp truncated_hash("0x000000000000000000000000" <> rest) do
+    "0x" <> rest
+  end
+end

apps/indexer/test/support/indexer/token_fetcher_case.ex

@@ -0,0 +1,16 @@
+defmodule Indexer.TokenFetcherCase do
+  alias Indexer.TokenFetcher
+
+  def start_supervised!(options \\ []) when is_list(options) do
+    options
+    |> Keyword.merge(
+      flush_interval: 50,
+      init_chunk_size: 1,
+      max_batch_size: 1,
+      max_concurrency: 1,
+      name: TokenFetcher
+    )
+    |> TokenFetcher.child_spec()
+    |> ExUnit.Callbacks.start_supervised!()
+  end
+end

mix.lock

@@ -21,9 +21,9 @@
   "deep_merge": {:hex, :deep_merge, "0.1.1", "c27866a7524a337b6a039eeb8dd4f17d458fd40fbbcb8c54661b71a22fffe846", [:mix], [], "hexpm"},
   "dialyxir": {:hex, :dialyxir, "0.5.1", "b331b091720fd93e878137add264bac4f644e1ddae07a70bf7062c7862c4b952", [:mix], [], "hexpm"},
   "earmark": {:hex, :earmark, "1.2.5", "4d21980d5d2862a2e13ec3c49ad9ad783ffc7ca5769cf6ff891a4553fbaae761", [:mix], [], "hexpm"},
-  "ecto": {:hex, :ecto, "2.2.8", "a4463c0928b970f2cee722cd29aaac154e866a15882c5737e0038bbfcf03ec2c", [:mix], [{:db_connection, "~> 1.1", [hex: :db_connection, repo: "hexpm", optional: true]}, {:decimal, "~> 1.2", [hex: :decimal, repo: "hexpm", optional: false]}, {:mariaex, "~> 0.8.0", [hex: :mariaex, repo: "hexpm", optional: true]}, {:poison, "~> 2.2 or ~> 3.0", [hex: :poison, repo: "hexpm", optional: true]}, {:poolboy, "~> 1.5", [hex: :poolboy, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.13.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:sbroker, "~> 1.0", [hex: :sbroker, repo: "hexpm", optional: true]}], "hexpm"},
+  "ecto": {:hex, :ecto, "2.2.10", "e7366dc82f48f8dd78fcbf3ab50985ceeb11cb3dc93435147c6e13f2cda0992e", [:mix], [{:db_connection, "~> 1.1", [hex: :db_connection, repo: "hexpm", optional: true]}, {:decimal, "~> 1.2", [hex: :decimal, repo: "hexpm", optional: false]}, {:mariaex, "~> 0.8.0", [hex: :mariaex, repo: "hexpm", optional: true]}, {:poison, "~> 2.2 or ~> 3.0", [hex: :poison, repo: "hexpm", optional: true]}, {:poolboy, "~> 1.5", [hex: :poolboy, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.13.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:sbroker, "~> 1.0", [hex: :sbroker, repo: "hexpm", optional: true]}], "hexpm"},
   "elixir_make": {:hex, :elixir_make, "0.4.1", "6628b86053190a80b9072382bb9756a6c78624f208ec0ff22cb94c8977d80060", [:mix], [], "hexpm"},
-  "ex_abi": {:hex, :ex_abi, "0.1.13", "717d88fac1d849774d28179aa958e4e55027793934c4dd0d25aa739720716a13", [:mix], [{:exth_crypto, "~> 0.1.4", [hex: :exth_crypto, repo: "hexpm", optional: false]}], "hexpm"},
+  "ex_abi": {:hex, :ex_abi, "0.1.14", "425eb3dacbc284a907acdd79dd0657a66794bc823ee8ceae06e72f7b997ef48c", [:mix], [{:exth_crypto, "~> 0.1.4", [hex: :exth_crypto, repo: "hexpm", optional: false]}], "hexpm"},
   "ex_cldr": {:hex, :ex_cldr, "1.3.2", "8f4a00c99d1c537b8e8db7e7903f4bd78d82a7289502d080f70365392b13921b", [:mix], [{:abnf2, "~> 0.1", [hex: :abnf2, repo: "hexpm", optional: false]}, {:decimal, "~> 1.4", [hex: :decimal, repo: "hexpm", optional: false]}, {:gettext, "~> 0.13", [hex: :gettext, repo: "hexpm", optional: true]}, {:poison, "~> 2.1 or ~> 3.0", [hex: :poison, repo: "hexpm", optional: true]}], "hexpm"},
   "ex_cldr_numbers": {:hex, :ex_cldr_numbers, "1.2.0", "ef27299922da913ffad1ed296cacf28b6452fc1243b77301dc17c03276c6ee34", [:mix], [{:decimal, "~> 1.4", [hex: :decimal, repo: "hexpm", optional: false]}, {:ex_cldr, "~> 1.3", [hex: :ex_cldr, repo: "hexpm", optional: false]}, {:poison, "~> 2.1 or ~> 3.1", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm"},
   "ex_cldr_units": {:hex, :ex_cldr_units, "1.1.1", "b3c7256709bdeb3740a5f64ce2bce659eb9cf4cc1afb4cf94aba033b4a18bc5f", [:mix], [{:ex_cldr, "~> 1.0", [hex: :ex_cldr, repo: "hexpm", optional: false]}, {:ex_cldr_numbers, "~> 1.0", [hex: :ex_cldr_numbers, repo: "hexpm", optional: false]}], "hexpm"},
@@ -55,7 +55,7 @@
   "mimerl": {:hex, :mimerl, "1.0.2", "993f9b0e084083405ed8252b99460c4f0563e41729ab42d9074fd5e52439be88", [:rebar3], [], "hexpm"},
   "mochiweb": {:hex, :mochiweb, "2.15.0", "e1daac474df07651e5d17cc1e642c4069c7850dc4508d3db7263a0651330aacc", [:rebar3], [], "hexpm"},
   "mock": {:hex, :mock, "0.3.1", "994f00150f79a0ea50dc9d86134cd9ebd0d177ad60bd04d1e46336cdfdb98ff9", [:mix], [{:meck, "~> 0.8.8", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm"},
-  "mox": {:hex, :mox, "0.3.2", "3b9b8364fd4f28628139de701d97c636b27a8f925f57a8d5a1b85fbd620dad3a", [:mix], [], "hexpm"},
+  "mox": {:hex, :mox, "0.4.0", "7f120840f7d626184a3d65de36189ca6f37d432e5d63acd80045198e4c5f7e6e", [:mix], [], "hexpm"},
   "parallel_stream": {:hex, :parallel_stream, "1.0.6", "b967be2b23f0f6787fab7ed681b4c45a215a81481fb62b01a5b750fa8f30f76c", [:mix], [], "hexpm"},
   "parse_trans": {:hex, :parse_trans, "3.2.0", "2adfa4daf80c14dc36f522cf190eb5c4ee3e28008fc6394397c16f62a26258c2", [:rebar3], [], "hexpm"},
   "phoenix": {:hex, :phoenix, "1.3.0", "1c01124caa1b4a7af46f2050ff11b267baa3edb441b45dbf243e979cd4c5891b", [:mix], [{:cowboy, "~> 1.0", [hex: :cowboy, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 1.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:plug, "~> 1.3.3 or ~> 1.4", [hex: :plug, repo: "hexpm", optional: false]}, {:poison, "~> 2.2 or ~> 3.0", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm"},