use or_where statement to update replaced transactions

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

@@ -50,8 +50,8 @@ defmodule Explorer.Chain.Import.Runner.Transactions do
     |> Multi.run(:transactions, fn repo, _ ->
       insert(repo, changes_list, insert_options)
     end)
-    |> Multi.run(:replaced_transactions, fn repo, %{transactions: transactions} ->
+    |> Multi.run(:replaced_transactions, fn repo, _ ->
-      update_replaced_transactions(repo, transactions, update_transactions_options)
+      update_replaced_transactions(repo, changes_list, update_transactions_options)
     end)
   end
 
@@ -188,38 +188,37 @@ defmodule Explorer.Chain.Import.Runner.Transactions do
   defp put_internal_transactions_indexed_at?(_, _), do: false
 
   defp update_replaced_transactions(repo, transactions, %{timeout: timeout}) do
-    result =
+    filters =
       transactions
-      |> Enum.filter(& &1.block_hash)
+      |> Enum.filter(fn transaction ->
-      |> Enum.map(fn transaction -> {transaction.nonce, transaction.from_address_hash} end)
+        transaction.block_hash && transaction.block_number && transaction.nonce && transaction.from_address_hash
-      |> Enum.uniq()
-      |> Enum.map(fn {nonce, from_address_hash} ->
-        from(t in Transaction,
-          where: t.nonce == ^nonce and t.from_address_hash == ^from_address_hash and is_nil(t.block_hash),
-          update: [
-            set: [status: ^:error, error: "dropped/replaced"]
-          ]
-        )
       end)
-      |> Enum.map(fn query ->
+      |> Enum.map(fn transaction ->
-        try do
+        {transaction.nonce, transaction.from_address_hash}
-          {_, result} = repo.update_all(query, [], timeout: timeout)
-
-          {:ok, result}
-        rescue
-          postgrex_error in Postgrex.Error ->
-            {:error, %{exception: postgrex_error, query: query}}
-        end
-      end)
-
-    first_error =
-      Enum.find(result, fn result ->
-        case result do
-          {:ok, _} -> false
-          {:error, _} -> true
-        end
       end)
+      |> Enum.uniq()
 
-    if first_error, do: first_error, else: {:ok, []}
+    if Enum.empty?(filters) do
+      {:ok, []}
+    else
+      query =
+        filters
+        |> Enum.reduce(from(t in Transaction, where: is_nil(t.block_hash)), fn {nonce, from_address}, query ->
+          from(t in query,
+            or_where: t.nonce == ^nonce and t.from_address_hash == ^from_address and is_nil(t.block_hash)
+          )
+        end)
+
+      update_query = from(t in query, update: [set: [status: ^:error, error: "dropped/replaced"]])
+
+      try do
+        {_, result} = repo.update_all(update_query, [], timeout: timeout)
+
+        {:ok, result}
+      rescue
+        postgrex_error in Postgrex.Error ->
+          {:error, %{exception: postgrex_error, query: query}}
+      end
+    end
   end
 end

apps/explorer/test/explorer/chain/import_test.exs

@@ -1035,6 +1035,92 @@ defmodule Explorer.Chain.ImportTest do
                Repo.get(Transaction, "0xab349efbe1ddc6d85d84a993aa52bdaadce66e8ee166dd10013ce3f2a94ca724")
     end
 
+    test "updates replaced transaction's status and error message" do
+      replaced_transaction_hash = "0x2a263224a95275d77bc30a7e131bc64d948777946a790c0915ab293791fbcb61"
+
+      address = insert(:address, hash: "0xb7cffe2ac19b9d5705a24cbe14fef5663af905a6")
+
+      insert(:transaction,
+        from_address: address,
+        nonce: 1,
+        block_hash: nil,
+        index: nil,
+        block_number: nil,
+        hash: replaced_transaction_hash
+      )
+
+      assert {:ok, _} =
+               Import.all(%{
+                 addresses: %{
+                   params: [
+                     %{hash: "0x1c0fa194a9d3b44313dcd849f3c6be6ad270a0a4"},
+                     %{hash: "0x679ed2245eba484021c2d3f4d174fb2bb2bd0e49"},
+                     %{hash: "0xb7cffe2ac19b9d5705a24cbe14fef5663af905a6"},
+                     %{hash: "0xfa52274dd61e1643d2205169732f29114bc240b3"}
+                   ]
+                 },
+                 address_coin_balances: %{
+                   params: [
+                     %{
+                       address_hash: "0xb7cffe2ac19b9d5705a24cbe14fef5663af905a6",
+                       block_number: 6_535_159
+                     }
+                   ]
+                 },
+                 blocks: %{
+                   params: [
+                     %{
+                       consensus: true,
+                       difficulty: 242_354_495_292_210,
+                       gas_limit: 4_703_218,
+                       gas_used: 1_009_480,
+                       hash: "0x1f8cde8bd326702c49e065d56b08bdc82caa0c4820d914e27026c9c68ca1cf09",
+                       miner_hash: "0x1c0fa194a9d3b44313dcd849f3c6be6ad270a0a4",
+                       nonce: "0xafa5fc5c07f55ba5",
+                       number: 6_535_159,
+                       parent_hash: "0xd2cf6cf7a3d5455f450a2a3701995a7ad51f12010674883a6690cee337f75ffa",
+                       size: 4052,
+                       timestamp: DateTime.from_iso8601("2018-09-10 21:34:39Z") |> elem(1),
+                       total_difficulty: 415_641_295_487_918_824_165
+                     }
+                   ]
+                 },
+                 broadcast: false,
+                 transactions: %{
+                   params: [
+                     %{
+                       block_hash: "0x1f8cde8bd326702c49e065d56b08bdc82caa0c4820d914e27026c9c68ca1cf09",
+                       block_number: 6_535_159,
+                       cumulative_gas_used: 978_227,
+                       from_address_hash: "0xb7cffe2ac19b9d5705a24cbe14fef5663af905a6",
+                       gas: 978_227,
+                       gas_price: 99_000_000_000,
+                       gas_used: 978_227,
+                       hash: "0x1a263224a95275d77bc30a7e131bc64d948777946a790c0915ab293791fbcb61",
+                       index: 0,
+                       input: "0x",
+                       nonce: 1,
+                       r:
+                         33_589_694_337_999_451_585_110_289_972_555_130_664_768_096_048_542_148_916_928_040_955_524_640_045_158,
+                       s:
+                         42_310_749_137_599_445_408_044_732_541_966_181_996_695_356_587_068_481_874_121_265_172_051_825_560_665,
+                       status: nil,
+                       to_address_hash: nil,
+                       transaction_hash: "0x1a263224a95275d77bc30a7e131bc64d948777946a790c0915ab293791fbcb61",
+                       transaction_index: 0,
+                       v: 158,
+                       value: 0
+                     }
+                   ]
+                 }
+               })
+
+      assert %Transaction{status: nil, error: nil} =
+               Repo.get(Transaction, "0x1a263224a95275d77bc30a7e131bc64d948777946a790c0915ab293791fbcb61")
+
+      assert %Transaction{status: :error, error: "dropped/replaced"} = Repo.get(Transaction, replaced_transaction_hash)
+    end
+
     test "uncles record their transaction indexes in transactions_forks" do
       miner_hash = address_hash()
       from_address_hash = address_hash()