Live updates for internal_transactions on address page

Co-authored-by: Tim Mecklem <timothy@mecklem.com>

apps/block_scout_web/assets/js/pages/address.js

@@ -14,6 +14,7 @@ export const initialState = {
   beyondPageOne: null,
   channelDisconnected: false,
   filter: null,
+  newInternalTransactions: [],
   newTransactions: [],
   balance: null,
   transactionCount: null
@@ -42,6 +43,29 @@ export function reducer (state = initialState, action) {
         balance: action.msg.balance
       })
     }
+    case 'RECEIVED_NEW_INTERNAL_TRANSACTION_BATCH': {
+      if (state.channelDisconnected || state.beyondPageOne) return state
+
+      const incomingInternalTransactions = humps.camelizeKeys(action.msgs)
+        .filter(({toAddressHash, fromAddressHash}) => (
+          !state.filter ||
+          (state.filter === 'to' && toAddressHash === state.addressHash) ||
+          (state.filter === 'from' && fromAddressHash === state.addressHash)
+        ))
+
+      if (!state.batchCountAccumulator && action.msgs.length < BATCH_THRESHOLD) {
+        return Object.assign({}, state, {
+          newInternalTransactions: [
+            ...state.newInternalTransactions,
+            ...incomingInternalTransactions.map(({internalTransactionHtml}) => internalTransactionHtml)
+          ]
+        })
+      } else {
+        return Object.assign({}, state, {
+          batchCountAccumulator: state.batchCountAccumulator + action.msgs.length
+        })
+      }
+    }
     case 'RECEIVED_NEW_TRANSACTION_BATCH': {
       if (state.channelDisconnected || state.beyondPageOne) return state
 
@@ -75,26 +99,38 @@ export function reducer (state = initialState, action) {
 router.when('/address/:addressHash').then((params) => initRedux(reducer, {
   main (store) {
     const { addressHash, blockNumber } = params
-    const channel = socket.channel(`addresses:${addressHash}`, {})
+    const addressChannel = socket.channel(`addresses:${addressHash}`, {})
     store.dispatch({
       type: 'PAGE_LOAD',
       params,
       transactionCount: $('[data-selector="transaction-count"]').text()
     })
-    channel.join()
+    addressChannel.join()
-    channel.onError(() => store.dispatch({ type: 'CHANNEL_DISCONNECTED' }))
+    addressChannel.onError(() => store.dispatch({ type: 'CHANNEL_DISCONNECTED' }))
-    channel.on('balance', (msg) => store.dispatch({ type: 'RECEIVED_UPDATED_BALANCE', msg }))
+    addressChannel.on('balance', (msg) => store.dispatch({ type: 'RECEIVED_UPDATED_BALANCE', msg }))
-    if (!blockNumber) channel.on('transaction', batchChannel((msgs) => store.dispatch({ type: 'RECEIVED_NEW_TRANSACTION_BATCH', msgs })))
+    if (!blockNumber) {
+      addressChannel.on('transaction', batchChannel((msgs) =>
+        store.dispatch({ type: 'RECEIVED_NEW_TRANSACTION_BATCH', msgs })
+      ))
+    }
+    if (!blockNumber) {
+      addressChannel.on('internal_transaction', batchChannel((msgs) =>
+        store.dispatch({ type: 'RECEIVED_NEW_INTERNAL_TRANSACTION_BATCH', msgs })
+      ))
+    }
   },
   render (state, oldState) {
+    const $balance = $('[data-selector="balance-card"]')
     const $channelBatching = $('[data-selector="channel-batching-message"]')
     const $channelBatchingCount = $('[data-selector="channel-batching-count"]')
     const $channelDisconnected = $('[data-selector="channel-disconnected-message"]')
+    const $emptyInternalTransactionsList = $('[data-selector="empty-internal-transactions-list"]')
     const $emptyTransactionsList = $('[data-selector="empty-transactions-list"]')
-    const $balance = $('[data-selector="balance-card"]')
+    const $internalTransactionsList = $('[data-selector="internal-transactions-list"]')
     const $transactionCount = $('[data-selector="transaction-count"]')
     const $transactionsList = $('[data-selector="transactions-list"]')
 
+    if ($emptyInternalTransactionsList.length && state.newInternalTransactions.length) window.location.reload()
     if ($emptyTransactionsList.length && state.newTransactions.length) window.location.reload()
     if (state.channelDisconnected) $channelDisconnected.show()
     if (oldState.balance !== state.balance) $balance.empty().append(state.balance)
@@ -105,6 +141,9 @@ router.when('/address/:addressHash').then((params) => initRedux(reducer, {
     } else {
       $channelBatching.hide()
     }
+    if (oldState.newInternalTransactions !== state.newInternalTransactions && $internalTransactionsList.length) {
+      $internalTransactionsList.prepend(state.newInternalTransactions.slice(oldState.newInternalTransactions.length).reverse().join(''))
+    }
     if (oldState.newTransactions !== state.newTransactions && $transactionsList.length) {
       $transactionsList.prepend(state.newTransactions.slice(oldState.newTransactions.length).reverse().join(''))
       updateAllAges()

apps/block_scout_web/lib/block_scout_web/channels/address_channel.ex

@@ -4,10 +4,10 @@ defmodule BlockScoutWeb.AddressChannel do
   """
   use BlockScoutWeb, :channel
 
-  alias BlockScoutWeb.{AddressTransactionView, AddressView}
+  alias BlockScoutWeb.{AddressInternalTransactionView, AddressTransactionView, AddressView}
   alias Phoenix.View
 
-  intercept(["balance_update", "count", "transaction"])
+  intercept(["balance_update", "count", "internal_transaction", "transaction"])
 
   def join("addresses:" <> _address_hash, _params, socket) do
     {:ok, %{}, socket}
@@ -40,6 +40,27 @@ defmodule BlockScoutWeb.AddressChannel do
     {:noreply, socket}
   end
 
+  def handle_out("internal_transaction", %{address: address, internal_transaction: internal_transaction}, socket) do
+    Gettext.put_locale(BlockScoutWeb.Gettext, socket.assigns.locale)
+
+    rendered_internal_transaction =
+      View.render_to_string(
+        AddressInternalTransactionView,
+        "_internal_transaction.html",
+        locale: socket.assigns.locale,
+        address: address,
+        internal_transaction: internal_transaction
+      )
+
+    push(socket, "internal_transaction", %{
+      to_address_hash: to_string(internal_transaction.to_address_hash),
+      from_address_hash: to_string(internal_transaction.from_address_hash),
+      internal_transaction_html: rendered_internal_transaction
+    })
+
+    {:noreply, socket}
+  end
+
   def handle_out("transaction", %{address: address, transaction: transaction}, socket) do
     Gettext.put_locale(BlockScoutWeb.Gettext, socket.assigns.locale)
 

apps/block_scout_web/lib/block_scout_web/event_handler.ex

@@ -19,6 +19,7 @@ defmodule BlockScoutWeb.EventHandler do
     Chain.subscribe_to_events(:addresses)
     Chain.subscribe_to_events(:blocks)
     Chain.subscribe_to_events(:exchange_rate)
+    Chain.subscribe_to_events(:internal_transactions)
     Chain.subscribe_to_events(:transactions)
     {:ok, []}
   end

apps/block_scout_web/lib/block_scout_web/notifier.ex

@@ -36,6 +36,10 @@ defmodule BlockScoutWeb.Notifier do
     })
   end
 
+  def handle_event({:chain_event, :internal_transactions, internal_transactions}) do
+    Enum.each(internal_transactions, &broadcast_internal_transaction/1)
+  end
+
   def handle_event({:chain_event, :transactions, transaction_hashes}) do
     transaction_hashes
     |> Chain.hashes_to_transactions(
@@ -65,6 +69,24 @@ defmodule BlockScoutWeb.Notifier do
     })
   end
 
+  defp broadcast_internal_transaction(internal_transaction) do
+    Endpoint.broadcast("internal_transactions:new_internal_transaction", "new_internal_transaction", %{
+      internal_transaction: internal_transaction
+    })
+
+    Endpoint.broadcast("addresses:#{internal_transaction.from_address_hash}", "internal_transaction", %{
+      address: internal_transaction.from_address,
+      internal_transaction: internal_transaction
+    })
+
+    if internal_transaction.to_address_hash != internal_transaction.from_address_hash do
+      Endpoint.broadcast("addresses:#{internal_transaction.to_address_hash}", "internal_transaction", %{
+        address: internal_transaction.to_address,
+        internal_transaction: internal_transaction
+      })
+    end
+  end
+
   defp broadcast_transaction(transaction) do
     Endpoint.broadcast("transactions:new_transaction", "new_transaction", %{
       transaction: transaction
@@ -75,7 +97,7 @@ defmodule BlockScoutWeb.Notifier do
       transaction: transaction
     })
 
-    if transaction.to_address_hash && transaction.to_address_hash != transaction.from_address_hash do
+    if transaction.to_address_hash != transaction.from_address_hash do
       Endpoint.broadcast("addresses:#{transaction.to_address_hash}", "transaction", %{
         address: transaction.to_address,
         transaction: transaction

apps/block_scout_web/lib/block_scout_web/templates/address_internal_transaction/index.html.eex

@@ -116,12 +116,14 @@
         </div>
         <h2 class="card-title"><%= gettext "Internal Transactions" %></h2>
         <%= if Enum.count(@internal_transactions) > 0 do %>
-          <%= for internal_transaction <- @internal_transactions do %>
+          <span data-selector="internal-transactions-list">
-            <%= render "_internal_transaction.html", address: @address, internal_transaction: internal_transaction %>
+            <%= for internal_transaction <- @internal_transactions do %>
-          <% end %>
+              <%= render "_internal_transaction.html", address: @address, internal_transaction: internal_transaction %>
+            <% end %>
+          </span>
         <% else %>
           <div class="tile tile-muted text-center">
-            <span><%= gettext "There are no internal transactions for this address." %></span>
+            <span data-selector="empty-internal-transactions-list"><%= gettext "There are no internal transactions for this address." %></span>
           </div>
         <% end %>
         <div>

apps/block_scout_web/test/block_scout_web/channels/address_channel_test.exs

@@ -110,5 +110,75 @@ defmodule BlockScoutWeb.AddressChannelTest do
         100 -> assert true
       end
     end
+
+    test "notified of new_internal_transaction for matching from_address", %{address: address, topic: topic} do
+      transaction =
+        :transaction
+        |> insert(from_address: address)
+        |> with_block()
+
+      internal_transaction = insert(:internal_transaction, transaction: transaction, from_address: address, index: 0)
+
+      Notifier.handle_event({:chain_event, :internal_transactions, [internal_transaction]})
+
+      receive do
+        %Phoenix.Socket.Broadcast{topic: ^topic, event: "internal_transaction", payload: payload} ->
+          assert payload.address.hash == address.hash
+          assert payload.internal_transaction.id == internal_transaction.id
+      after
+        5_000 ->
+          assert false, "Expected message received nothing."
+      end
+    end
+
+    test "notified of new_internal_transaction for matching to_address", %{address: address, topic: topic} do
+      transaction =
+        :transaction
+        |> insert(to_address: address)
+        |> with_block()
+
+      internal_transaction = insert(:internal_transaction, transaction: transaction, to_address: address, index: 0)
+
+      Notifier.handle_event({:chain_event, :internal_transactions, [internal_transaction]})
+
+      receive do
+        %Phoenix.Socket.Broadcast{topic: ^topic, event: "internal_transaction", payload: payload} ->
+          assert payload.address.hash == address.hash
+          assert payload.internal_transaction.id == internal_transaction.id
+      after
+        5_000 ->
+          assert false, "Expected message received nothing."
+      end
+    end
+
+    test "not notified twice of new_internal_transaction if to and from address are equal", %{
+      address: address,
+      topic: topic
+    } do
+      transaction =
+        :transaction
+        |> insert(from_address: address, to_address: address)
+        |> with_block()
+
+      internal_transaction =
+        insert(:internal_transaction, transaction: transaction, from_address: address, to_address: address, index: 0)
+
+      Notifier.handle_event({:chain_event, :internal_transactions, [internal_transaction]})
+
+      receive do
+        %Phoenix.Socket.Broadcast{topic: ^topic, event: "internal_transaction", payload: payload} ->
+          assert payload.address.hash == address.hash
+          assert payload.internal_transaction.id == internal_transaction.id
+      after
+        5_000 ->
+          assert false, "Expected message received nothing."
+      end
+
+      receive do
+        _ -> assert false, "Received duplicate broadcast."
+      after
+        100 -> assert true
+      end
+    end
   end
 end

apps/block_scout_web/test/block_scout_web/features/pages/address_page.ex

@@ -31,6 +31,10 @@ defmodule BlockScoutWeb.AddressPage do
     css("[data-test='address_detail_hash']", text: to_string(address_hash))
   end
 
+  def internal_transaction(%InternalTransaction{id: id}) do
+    css("[data-test='internal_transaction'][data-internal-transaction-id='#{id}']")
+  end
+
   def internal_transactions(count: count) do
     css("[data-test='internal_transaction']", count: count)
   end

apps/block_scout_web/test/block_scout_web/features/viewing_addresses_test.exs

@@ -3,7 +3,7 @@ defmodule BlockScoutWeb.ViewingAddressesTest do
 
   alias Explorer.Chain.Wei
   alias Explorer.Factory
-  alias BlockScoutWeb.{AddressPage, AddressView}
+  alias BlockScoutWeb.{AddressPage, AddressView, Notifier}
 
   setup do
     block = insert(:block)
@@ -217,6 +217,27 @@ defmodule BlockScoutWeb.ViewingAddressesTest do
       |> AddressPage.click_internal_transactions()
       |> assert_has(AddressPage.internal_transaction_address_link(internal_transaction, :from))
     end
+
+    test "viewing new internal transactions via live update", %{addresses: addresses, session: session} do
+      transaction =
+        :transaction
+        |> insert(from_address: addresses.lincoln)
+        |> with_block()
+
+      session
+      |> AddressPage.visit_page(addresses.lincoln)
+      |> AddressPage.click_internal_transactions()
+      |> assert_has(AddressPage.internal_transactions(count: 2))
+
+      internal_transaction =
+        insert(:internal_transaction, transaction: transaction, index: 0, from_address: addresses.lincoln)
+
+      Notifier.handle_event({:chain_event, :internal_transactions, [internal_transaction]})
+
+      session
+      |> assert_has(AddressPage.internal_transactions(count: 3))
+      |> assert_has(AddressPage.internal_transaction(internal_transaction))
+    end
   end
 
   test "viewing transaction count", %{addresses: addresses, session: session} do

apps/explorer/lib/explorer/chain.ex

@@ -48,7 +48,8 @@ defmodule Explorer.Chain do
   @typedoc """
   Event type where data is broadcasted whenever data is inserted from chain indexing.
   """
-  @type chain_event :: :addresses | :balances | :blocks | :exchange_rate | :logs | :transactions
+  @type chain_event ::
+          :addresses | :balances | :blocks | :exchange_rate | :internal_transactions | :logs | :transactions
 
   @type direction :: :from | :to
 
@@ -1212,7 +1213,7 @@ defmodule Explorer.Chain do
   """
   @spec subscribe_to_events(chain_event()) :: :ok
   def subscribe_to_events(event_type)
-      when event_type in ~w(addresses balances blocks exchange_rate logs transactions)a do
+      when event_type in ~w(addresses balances blocks exchange_rate internal_transactions logs transactions)a do
     Registry.register(Registry.ChainEvents, event_type, [])
     :ok
   end

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

@@ -209,7 +209,8 @@ defmodule Explorer.Chain.Import do
   end
 
   defp broadcast_events(data) do
-    for {event_type, event_data} <- data, event_type in ~w(addresses balances blocks logs transactions)a do
+    for {event_type, event_data} <- data,
+        event_type in ~w(addresses balances blocks internal_transactions logs transactions)a do
       broadcast_event_data(event_type, event_data)
     end
   end
@@ -671,11 +672,7 @@ defmodule Explorer.Chain.Import do
         timestamps: timestamps
       )
 
-    {:ok,
+    {:ok, internal_transactions}
-     for(
-       internal_transaction <- internal_transactions,
-       do: Map.take(internal_transaction, [:index, :transaction_hash])
-     )}
   end
 
   @spec insert_logs([map()], %{required(:timeout) => timeout, required(:timestamps) => timestamps}) ::

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

@@ -11,6 +11,7 @@ defmodule Explorer.Chain.ImportTest do
     Log,
     Hash,
     Import,
+    InternalTransaction,
     Token,
     TokenTransfer,
     Transaction
@@ -417,6 +418,12 @@ defmodule Explorer.Chain.ImportTest do
       assert_received {:chain_event, :blocks, [%Block{}]}
     end
 
+    test "publishes internal_transaction data to subscribers on insert" do
+      Chain.subscribe_to_events(:internal_transactions)
+      Import.all(@import_data)
+      assert_received {:chain_event, :internal_transactions, [%InternalTransaction{}]}
+    end
+
     test "publishes log data to subscribers on insert" do
       Chain.subscribe_to_events(:logs)
       Import.all(@import_data)