Delete block_rewards for non-consensus or old consensus blocks

Fixes #1308

1. If a block is non-consensus delete rewards for its hash as the hash
should no longer have rewards.
2. If a block is consensus delete rewards for its number as rewards
currently in the database were from before a reorg.

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

@@ -10,6 +10,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
   alias Ecto.Adapters.SQL
   alias Ecto.{Changeset, Multi, Repo}
   alias Explorer.Chain.{Block, Import, InternalTransaction, Transaction}
+  alias Explorer.Chain.Block.Reward
   alias Explorer.Chain.Import.Runner
 
   @behaviour Runner
@@ -65,6 +66,9 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     |> Multi.run(:lose_consenus, fn repo, _ ->
       lose_consensus(repo, changes_list, insert_options)
     end)
+    |> Multi.run(:remove_rewards, fn repo, _ ->
+      delete_rewards(repo, changes_list, insert_options)
+    end)
     |> Multi.run(:blocks, fn repo, _ ->
       insert(repo, changes_list, insert_options)
     end)
@@ -276,6 +280,34 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     end
   end
 
+  # `block_rewards` are linked to `blocks.hash`, but fetched by `blocks.number`, so when a block with the same number is
+  # inserted, the old block rewards need to be deleted, so that the old and new rewards aren't combined.
+  defp delete_rewards(repo, blocks_changes, %{timeout: timeout}) do
+    {hashes, numbers} =
+      Enum.reduce(blocks_changes, {[], []}, fn
+        %{consensus: false, hash: hash}, {acc_hashes, acc_numbers} ->
+          {[hash | acc_hashes], acc_numbers}
+
+        %{consensus: true, number: number}, {acc_hashes, acc_numbers} ->
+          {acc_hashes, [number | acc_numbers]}
+      end)
+
+    query =
+      from(reward in Reward,
+        inner_join: block in assoc(reward, :block),
+        where: block.hash in ^hashes or block.number in ^numbers
+      )
+
+    try do
+      {count, nil} = repo.delete_all(query, timeout: timeout)
+
+      {:ok, count}
+    rescue
+      postgrex_error in Postgrex.Error ->
+        {:error, %{exception: postgrex_error, blocks_changes: blocks_changes}}
+    end
+  end
+
   defp update_block_second_degree_relations(repo, blocks, %{timeout: timeout, timestamps: %{updated_at: updated_at}})
        when is_list(blocks) do
     ordered_uncle_hashes =