Add chain function to get cataloged tokens

apps/explorer/lib/explorer/chain.ex

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

apps/explorer/test/explorer/chain_test.exs

@@ -2756,6 +2756,12 @@ defmodule Explorer.ChainTest do
     assert Chain.stream_uncataloged_token_contract_address_hashes([], &[&1 | &2]) == {:ok, [uncatalog_address]}
   end
 
+  test "stream_cataloged_token_contract_address_hashes/2 reduces with given reducer and accumulator" do
+    %Token{contract_address_hash: catalog_address} = insert(:token, cataloged: true)
+    insert(:token, cataloged: false)
+    assert Chain.stream_cataloged_token_contract_address_hashes([], &[&1 | &2]) == {:ok, [catalog_address]}
+  end
+
   describe "transaction_has_token_transfers?/1" do
     test "returns true if transaction has token transfers" do
       transaction = insert(:transaction)