Explorer.Chain.stream_unfetched_uncle_hashes

Streams unique uncle hashes from block_second_degree_relations.

apps/explorer/lib/explorer/chain.ex

@@ -932,6 +932,37 @@ defmodule Explorer.Chain do
     )
   end
 
+  @doc """
+  Returns a stream of all `t:Explorer.Chain.Block.t/0` `hash`es that are marked as unfetched in
+  `t:Explorer.Chain.Block.SecondDegreeRelation.t/0`.
+
+  When a block is fetched, its uncles are transformed into `t:Explorer.Chain.Block.SecondDegreeRelation.t/0` and can be
+  returned.  Once the uncle is imported its corresponding `t:Explorer.Chain.Block.SecondDegreeRelation.t/0`
+  `uncle_fetched_at` will be set and it won't be returned anymore.
+  """
+  @spec stream_unfetched_uncle_hashes(
+          initial :: accumulator,
+          reducer :: (entry :: Hash.Full.t(), accumulator -> accumulator)
+        ) :: {:ok, accumulator}
+        when accumulator: term()
+  def stream_unfetched_uncle_hashes(initial, reducer) when is_function(reducer, 2) do
+    Repo.transaction(
+      fn ->
+        query =
+          from(bsdr in Block.SecondDegreeRelation,
+            where: is_nil(bsdr.uncle_fetched_at),
+            select: bsdr.uncle_hash,
+            group_by: bsdr.uncle_hash
+          )
+
+        query
+        |> Repo.stream(timeout: :infinity)
+        |> Enum.reduce(initial, reducer)
+      end,
+      timeout: :infinity
+    )
+  end
+
   @doc """
   The number of `t:Explorer.Chain.Log.t/0`.
 

apps/explorer/test/explorer/chain_test.exs

@@ -2272,6 +2272,20 @@ defmodule Explorer.ChainTest do
     end
   end
 
+  describe "stream_unfetched_uncle_hashes/2" do
+    test "does not return uncle hashes where t:Explorer.Chain.Block.SecondDegreeRelation.t/0 unclue_fetched_at is not nil" do
+      %Block.SecondDegreeRelation{nephew: %Block{}, uncle_hash: uncle_hash} = insert(:block_second_degree_relation)
+
+      assert {:ok, [^uncle_hash]} = Explorer.Chain.stream_unfetched_uncle_hashes([], &[&1 | &2])
+
+      query = from(bsdr in Block.SecondDegreeRelation, where: bsdr.uncle_hash == ^uncle_hash)
+
+      assert {1, _} = Repo.update_all(query, set: [uncle_fetched_at: DateTime.utc_now()])
+
+      assert {:ok, []} = Explorer.Chain.stream_unfetched_uncle_hashes([], &[&1 | &2])
+    end
+  end
+
   test "total_supply/0" do
     height = 2_000_000
     insert(:block, number: height)