stream unfetched token instances

apps/explorer/lib/explorer/chain.ex

@@ -2823,6 +2823,27 @@ defmodule Explorer.Chain do
     Repo.stream_reduce(query, initial, reducer)
   end
 
+  @spec stream_unfetched_token_instances(
+          initial :: accumulator,
+          reducer :: (entry :: map(), accumulator -> accumulator)
+        ) :: {:ok, accumulator}
+        when accumulator: term()
+  def stream_unfetched_token_instances(initial, reducer) when is_function(reducer, 2) do
+    query =
+      from(
+        token_transfer in TokenTransfer,
+        inner_join: token in Token,
+        on: token.contract_address_hash == token_transfer.token_contract_address_hash,
+        left_join: instance in Instance,
+        on: token_transfer.token_id == instance.token_id,
+        where: token.type == ^"ERC-721" and is_nil(instance.token_id),
+        distinct: token_transfer.token_id,
+        select: %{contract_address_hash: token_transfer.token_contract_address_hash, token_id: token_transfer.token_id}
+      )
+
+    Repo.stream_reduce(query, initial, reducer)
+  end
+
   @doc """
   Streams a list of token contract addresses that have been cataloged.
   """

apps/explorer/priv/repo/migrations/20190905083522_create_token_instances.exs

@@ -16,5 +16,7 @@ defmodule Explorer.Repo.Migrations.CreateTokenInstances do
 
       timestamps(null: false, type: :utc_datetime_usec)
     end
+
+    create_if_not_exists(index(:token_instances, [:token_id]))
   end
 end

apps/explorer/test/explorer/chain_test.exs

@@ -3602,6 +3602,61 @@ defmodule Explorer.ChainTest do
     end
   end
 
+  describe "stream_unfetched_token_instances/2" do
+    test "reduces wuth given reducer and accumulator" do
+      token_contract_address = insert(:contract_address)
+      token = insert(:token, contract_address: token_contract_address, type: "ERC-721")
+
+      transaction =
+        :transaction
+        |> insert()
+        |> with_block(insert(:block, number: 1))
+
+      token_transfer =
+        insert(
+          :token_transfer,
+          block_number: 1000,
+          to_address: build(:address),
+          transaction: transaction,
+          token_contract_address: token_contract_address,
+          token: token,
+          token_id: 11
+        )
+
+      assert {:ok, [result]} = Chain.stream_unfetched_token_instances([], &[&1 | &2])
+      assert result.token_id == token_transfer.token_id
+      assert result.contract_address_hash == token_transfer.token_contract_address_hash
+    end
+
+    test "do not fetch records with token instances" do
+      token_contract_address = insert(:contract_address)
+      token = insert(:token, contract_address: token_contract_address, type: "ERC-721")
+
+      transaction =
+        :transaction
+        |> insert()
+        |> with_block(insert(:block, number: 1))
+
+      token_transfer =
+        insert(
+          :token_transfer,
+          block_number: 1000,
+          to_address: build(:address),
+          transaction: transaction,
+          token_contract_address: token_contract_address,
+          token: token,
+          token_id: 11
+        )
+
+      insert(:token_instance,
+        token_id: token_transfer.token_id,
+        token_contract_address_hash: token_transfer.token_contract_address_hash
+      )
+
+      assert {:ok, []} = Chain.stream_unfetched_token_instances([], &[&1 | &2])
+    end
+  end
+
   describe "search_token/1" do
     test "finds by part of the name" do
       token = insert(:token, name: "magic token", symbol: "MAGIC")

apps/explorer/test/support/factory.ex

@@ -26,6 +26,7 @@ defmodule Explorer.Factory do
     SmartContract,
     Token,
     TokenTransfer,
+    Token.Instance,
     Transaction,
     StakingPool,
     StakingPoolsDelegator
@@ -542,6 +543,14 @@ defmodule Explorer.Factory do
     }
   end
 
+  def token_instance_factory do
+    %Instance{
+      token_contract_address_hash: build(:address),
+      token_id: 5,
+      metadata: %{key: "value"}
+    }
+  end
+
   def token_balance_factory do
     %TokenBalance{
       address: build(:address),