Create worker for importing internal transactions

config/config.exs

@@ -41,7 +41,13 @@ config :exq,
   scheduler_enable: true,
   shutdown_timeout: 5000,
   max_retries: 10,
-  queues: [{"default", 1}, {"blocks", 1}, {"transactions", 1}, {"receipts", 1}]
+  queues: [
+    {"default", 1},
+    {"blocks", 1},
+    {"internal_transactions", 1},
+    {"transactions", 1},
+    {"receipts", 1}
+  ]
 
 config :exq_ui, server: false
 

config/prod.exs

@@ -82,6 +82,8 @@ config :exq,
   queues: [
     {"blocks", String.to_integer(System.get_env("EXQ_BLOCKS_CONCURRENCY") || "1")},
     {"default", String.to_integer(System.get_env("EXQ_CONCURRENCY") || "1")},
+    {"internal_transactions",
+     String.to_integer(System.get_env("EXQ_INTERNAL_TRANSACTIONS_CONCURRENCY") || "1")},
     {"receipts", String.to_integer(System.get_env("EXQ_RECEIPTS_CONCURRENCY") || "1")},
     {"transactions", String.to_integer(System.get_env("EXQ_TRANSACTIONS_CONCURRENCY") || "1")}
   ]

lib/explorer/workers/import_internal_transaction.ex

@@ -0,0 +1,12 @@
+defmodule Explorer.Workers.ImportInternalTransaction do
+  @moduledoc "Imports internal transactions via Parity trace endpoints."
+
+  alias Explorer.InternalTransactionImporter
+
+  @dialyzer {:nowarn_function, perform: 1}
+  def perform(hash), do: InternalTransactionImporter.import(hash)
+
+  def perform_later(hash) do
+    Exq.enqueue(Exq.Enqueuer, "internal_transactions", __MODULE__, [hash])
+  end
+end

lib/explorer/workers/import_transaction.ex

@@ -5,16 +5,19 @@ defmodule Explorer.Workers.ImportTransaction do
 
   alias Explorer.TransactionImporter
   alias Explorer.Workers.ImportReceipt
+  alias Explorer.Workers.ImportInternalTransaction
 
   @dialyzer {:nowarn_function, perform: 1}
   def perform(hash) when is_binary(hash) do
     TransactionImporter.import(hash)
+    ImportInternalTransaction.perform_later(hash)
     ImportReceipt.perform_later(hash)
   end
 
   @dialyzer {:nowarn_function, perform: 1}
   def perform(raw_transaction) when is_map(raw_transaction) do
     TransactionImporter.import(raw_transaction)
+    ImportInternalTransaction.perform_later(raw_transaction["hash"])
     ImportReceipt.perform_later(raw_transaction["hash"])
   end
 

test/explorer/workers/import_internal_transaction_test.exs

@@ -0,0 +1,35 @@
+defmodule Explorer.Workers.ImportInternalTransactionTest do
+  use Explorer.DataCase
+
+  alias Explorer.Repo
+  alias Explorer.InternalTransaction
+  alias Explorer.Workers.ImportInternalTransaction
+
+  describe "perform/1" do
+    test "does not import the internal transactions when no transaction with the hash exists" do
+      use_cassette "import_internal_transaction_perform_1" do
+        assert_raise Ecto.NoResultsError, fn ->
+          ImportInternalTransaction.perform(
+            "0xf9a0959d5ccde33ec5221ddba1c6d7eaf9580a8d3512c7a1a60301362a98f926"
+          )
+        end
+      end
+    end
+
+    test "imports a receipt when an internal transaction with the hash exists" do
+      insert(
+        :transaction,
+        hash: "0x051e031f05b3b3a5ff73e1189c36e3e2a41fd1c2d9772b2c75349e22ed4d3f68"
+      )
+
+      use_cassette "import_internal_transaction_perform_1" do
+        ImportInternalTransaction.perform(
+          "0x051e031f05b3b3a5ff73e1189c36e3e2a41fd1c2d9772b2c75349e22ed4d3f68"
+        )
+
+        internal_transaction_count = InternalTransaction |> Repo.all() |> Enum.count()
+        assert internal_transaction_count == 2
+      end
+    end
+  end
+end

test/explorer/workers/import_transaction_test.exs

@@ -3,9 +3,11 @@ defmodule Explorer.Workers.ImportTransactionTest do
 
   import Mock
 
+  alias Explorer.InternalTransaction
   alias Explorer.Receipt
   alias Explorer.Repo
   alias Explorer.Transaction
+  alias Explorer.Workers.ImportInternalTransaction
   alias Explorer.Workers.ImportTransaction
 
   describe "perform/1" do
@@ -51,12 +53,14 @@ defmodule Explorer.Workers.ImportTransactionTest do
 
       use_cassette "import_transaction_perform_1" do
         with_mock Exq, enqueue: fn _, _, _, _ -> insert(:receipt, transaction: transaction) end do
-          ImportTransaction.perform(
-            "0xf9a0959d5ccde33ec5221ddba1c6d7eaf9580a8d3512c7a1a60301362a98f926"
-          )
+          with_mock ImportInternalTransaction, perform_later: fn _ -> :ok end do
+            ImportTransaction.perform(
+              "0xf9a0959d5ccde33ec5221ddba1c6d7eaf9580a8d3512c7a1a60301362a98f926"
+            )
 
-          receipt = Repo.one(Receipt)
-          refute is_nil(receipt)
+            receipt = Repo.one(Receipt)
+            refute is_nil(receipt)
+          end
         end
       end
     end
@@ -70,15 +74,64 @@ defmodule Explorer.Workers.ImportTransactionTest do
 
       use_cassette "import_transaction_perform_1" do
         with_mock Exq, enqueue: fn _, _, _, _ -> insert(:receipt, transaction: transaction) end do
-          ImportTransaction.perform(%{
-            "hash" => "0xf9a0959d5ccde33ec5221ddba1c6d7eaf9580a8d3512c7a1a60301362a98f926",
-            "to" => "0xc001",
-            "from" => "0xbead5",
-            "blockHash" => "0xcafe"
-          })
-
-          receipt = Repo.one(Receipt)
-          refute is_nil(receipt)
+          with_mock ImportInternalTransaction, perform_later: fn _ -> :ok end do
+            ImportTransaction.perform(%{
+              "hash" => "0xf9a0959d5ccde33ec5221ddba1c6d7eaf9580a8d3512c7a1a60301362a98f926",
+              "to" => "0xc001",
+              "from" => "0xbead5",
+              "blockHash" => "0xcafe"
+            })
+
+            receipt = Repo.one(Receipt)
+            refute is_nil(receipt)
+          end
+        end
+      end
+    end
+
+    test "imports the internal transactions in another queue" do
+      transaction =
+        insert(
+          :transaction,
+          hash: "0xf9a0959d5ccde33ec5221ddba1c6d7eaf9580a8d3512c7a1a60301362a98f926"
+        )
+
+      use_cassette "import_transaction_perform_1" do
+        with_mock Exq, enqueue: fn _, _, _, _ -> :ok end do
+          with_mock ImportInternalTransaction,
+            perform_later: fn _ -> insert(:internal_transaction, transaction: transaction) end do
+            ImportTransaction.perform(
+              "0xf9a0959d5ccde33ec5221ddba1c6d7eaf9580a8d3512c7a1a60301362a98f926"
+            )
+
+            internal_transaction = Repo.one(InternalTransaction)
+            refute is_nil(internal_transaction)
+          end
+        end
+      end
+    end
+
+    test "imports the internal transactions in another queue when a map is supplied" do
+      transaction =
+        insert(
+          :transaction,
+          hash: "0xf9a0959d5ccde33ec5221ddba1c6d7eaf9580a8d3512c7a1a60301362a98f926"
+        )
+
+      use_cassette "import_transaction_perform_1" do
+        with_mock Exq, enqueue: fn _, _, _, _ -> :ok end do
+          with_mock ImportInternalTransaction,
+            perform_later: fn _ -> insert(:internal_transaction, transaction: transaction) end do
+            ImportTransaction.perform(%{
+              "hash" => "0xf9a0959d5ccde33ec5221ddba1c6d7eaf9580a8d3512c7a1a60301362a98f926",
+              "to" => "0xc001",
+              "from" => "0xbead5",
+              "blockHash" => "0xcafe"
+            })
+
+            internal_transaction = Repo.one(InternalTransaction)
+            refute is_nil(internal_transaction)
+          end
         end
       end
     end

test/support/factories/internal_transaction_factory.ex

@@ -8,7 +8,6 @@ defmodule Explorer.InternalTransactionFactory do
           trace_address: [Enum.random(0..4), Enum.random(0..4)],
           from_address_id: insert(:address).id,
           to_address_id: insert(:address).id,
-          transaction_id: insert(:transaction).id,
           value: Enum.random(1..100_000),
           gas: Enum.random(1..100_000),
           gas_used: Enum.random(1..100_000),

test/support/fixture/vcr_cassettes/import_internal_transaction_perform_1.json

@@ -0,0 +1 @@
+[]

test/support/fixture/vcr_cassettes/transaction_importer_binds_internal_transactions.json

@@ -0,0 +1,30 @@
+[
+  {
+    "request": {
+      "body": "{\"params\":[\"0x051e031f05b3b3a5ff73e1189c36e3e2a41fd1c2d9772b2c75349e22ed4d3f68\"],\"method\":\"eth_getTransactionByHash\",\"jsonrpc\":\"2.0\",\"id\":0}",
+      "headers": {
+        "Content-Type": "application/json"
+      },
+      "method": "post",
+      "options": [],
+      "request_body": "",
+      "url": "https://sokol-trace.poa.network"
+    },
+    "response": {
+      "binary": false,
+      "body": "{\"jsonrpc\":\"2.0\",\"result\":{\"blockHash\":\"0x764e83f5e0c2d0b8b8b28421b430267c7d02260caaf014495a7deb3550b051bb\",\"blockNumber\":\"0x94b87\",\"chainId\":\"0x4d\",\"condition\":null,\"creates\":null,\"from\":\"0x82e4e61e7f5139ff0a4157a5bc687ef42294c248\",\"gas\":\"0xb5e4\",\"gasPrice\":\"0x3b9aca00\",\"hash\":\"0x051e031f05b3b3a5ff73e1189c36e3e2a41fd1c2d9772b2c75349e22ed4d3f68\",\"input\":\"0x\",\"nonce\":\"0x190\",\"publicKey\":\"0x750f765a7790aa361dde0ff1d4c9dadd0e1c62fe79dee9c3d6b953abe99ce79755568215a2524d0efe0d5a48d8082e86f317434b62964b0c2e3e3446ebe08099\",\"r\":\"0x94d42f80a1137906cb3e8733ec12837e88d862edf1757de90d35ef0aa90cbc7a\",\"raw\":\"0xf86e820190843b9aca0082b5e4940e8a835861fe709b6100f83628debbc353342734880f95b28cd2c380008081bea094d42f80a1137906cb3e8733ec12837e88d862edf1757de90d35ef0aa90cbc7aa03da8b0651dec4d7f747177792e6023252db9c6f0a137c4149e9c994c053cf22a\",\"s\":\"0x3da8b0651dec4d7f747177792e6023252db9c6f0a137c4149e9c994c053cf22a\",\"standardV\":\"0x1\",\"to\":\"0x0e8a835861fe709b6100f83628debbc353342734\",\"transactionIndex\":\"0x0\",\"v\":\"0xbe\",\"value\":\"0xf95b28cd2c38000\"},\"id\":0}\n",
+      "headers": {
+        "Date": "Fri, 23 Feb 2018 22:45:24 GMT",
+        "Content-Type": "application/json",
+        "Transfer-Encoding": "chunked",
+        "Connection": "keep-alive",
+        "Set-Cookie": "__cfduid=d34a81fb10f3b5a70f78d46097ba3eedc1519425923; expires=Sat, 23-Feb-19 22:45:23 GMT; path=/; domain=.poa.network; HttpOnly; Secure",
+        "Expect-CT": "max-age=604800, report-uri=\"https://report-uri.cloudflare.com/cdn-cgi/beacon/expect-ct\"",
+        "Server": "cloudflare",
+        "CF-RAY": "3f1db7189cd26c04-SJC"
+      },
+      "status_code": 200,
+      "type": "ok"
+    }
+  }
+]