remove old internal transactions with the same primary key

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

@@ -71,13 +71,13 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactions do
                                                   } ->
       valid_internal_transactions(transactions, internal_transactions_params, invalid_block_numbers)
     end)
-    |> Multi.run(:internal_transactions, fn repo, %{valid_internal_transactions: valid_internal_transactions} ->
-      insert(repo, valid_internal_transactions, insert_options)
-    end)
     |> Multi.run(:remove_left_over_internal_transactions, fn repo,
                                                              %{valid_internal_transactions: valid_internal_transactions} ->
       remove_left_over_internal_transactions(repo, valid_internal_transactions)
     end)
+    |> Multi.run(:internal_transactions, fn repo, %{valid_internal_transactions: valid_internal_transactions} ->
+      insert(repo, valid_internal_transactions, insert_options)
+    end)
     |> Multi.run(:update_transactions, fn repo, %{valid_internal_transactions: valid_internal_transactions} ->
       update_transactions(repo, valid_internal_transactions, update_transactions_options)
     end)
@@ -289,7 +289,7 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactions do
 
       _ ->
         try do
-          delete_query =
+          delete_query_for_block_hash_block_index =
             valid_internal_transactions
             |> Enum.group_by(& &1.block_hash, & &1.block_index)
             |> Enum.map(fn {block_hash, indexes} -> {block_hash, Enum.max(indexes)} end)
@@ -297,10 +297,18 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactions do
               or_where(acc, [it], it.block_hash == ^block_hash and it.block_index > ^max_index)
             end)
 
+          # removes old recoreds with the same primary key (transaction hash, transaction index)
+          delete_query =
+            valid_internal_transactions
+            |> Enum.map(fn params -> {params.transaction_hash, params.index} end)
+            |> Enum.reduce(delete_query_for_block_hash_block_index, fn {transaction_hash, index}, acc ->
+              or_where(acc, [it], it.transaction_hash == ^transaction_hash and it.index == ^index)
+            end)
+
           # ShareLocks order already enforced by `acquire_pending_internal_txs` (see docs: sharelocks.md)
-          {_count, result} = repo.delete_all(delete_query, [])
+          {count, result} = repo.delete_all(delete_query, [])
 
-          {:ok, result}
+          {:ok, {count, result}}
         rescue
           postgrex_error in Postgrex.Error -> {:error, %{exception: postgrex_error}}
         end

apps/explorer/test/explorer/chain/import/runner/internal_transactions_test.exs

@@ -85,6 +85,33 @@ defmodule Explorer.Chain.Import.Runner.InternalTransactionsTest do
       assert PendingBlockOperation |> Repo.get(full_block.hash) |> is_nil()
     end
 
+    test "removes old records with the same primary key (transaction_hash, index)" do
+      full_block = insert(:block)
+      another_full_block = insert(:block)
+
+      transaction = insert(:transaction) |> with_block(full_block)
+
+      insert(:internal_transaction,
+        index: 0,
+        transaction: transaction,
+        block_hash: another_full_block.hash,
+        block_index: 0
+      )
+
+      insert(:pending_block_operation, block_hash: full_block.hash, fetch_internal_transactions: true)
+
+      transaction_changes = make_internal_transaction_changes(transaction, 0, nil)
+
+      assert {:ok, %{remove_left_over_internal_transactions: {1, nil}}} =
+               run_internal_transactions([transaction_changes])
+
+      assert from(i in InternalTransaction,
+               where: i.transaction_hash == ^transaction.hash and i.block_hash == ^another_full_block.hash
+             )
+             |> Repo.one()
+             |> is_nil()
+    end
+
     test "removes consensus to blocks where not all transactions are filled" do
       full_block = insert(:block)
       transaction_a = insert(:transaction) |> with_block(full_block)