pubsub for new token transfers

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## Current
 
 ### Features
+- [#3216](https://github.com/poanetwork/blockscout/pull/3216) - Display new token transfers at token page without refreshing the page
 - [#3199](https://github.com/poanetwork/blockscout/pull/3199) - Show compilation error at contract verification
 - [#3193](https://github.com/poanetwork/blockscout/pull/3193) - Raw trace copy button
 - [#3184](https://github.com/poanetwork/blockscout/pull/3184) - Apps navbar menu item

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

@@ -0,0 +1,87 @@
+import $ from 'jquery'
+import omit from 'lodash/omit'
+import URI from 'urijs'
+import humps from 'humps'
+import { subscribeChannel } from '../../socket'
+import { connectElements } from '../../lib/redux_helpers.js'
+import { createAsyncLoadStore } from '../../lib/async_listing_load'
+import '../address'
+
+export const initialState = {
+  addressHash: null,
+  channelDisconnected: false,
+  filter: null
+}
+
+export function reducer (state, action) {
+  switch (action.type) {
+    case 'PAGE_LOAD':
+    case 'ELEMENTS_LOAD': {
+      return Object.assign({}, state, omit(action, 'type'))
+    }
+    case 'CHANNEL_DISCONNECTED': {
+      if (state.beyondPageOne) return state
+
+      return Object.assign({}, state, { channelDisconnected: true })
+    }
+    case 'RECEIVED_NEW_TOKEN_TRANSFER': {
+      if (state.channelDisconnected) return state
+
+      if (state.beyondPageOne ||
+        (state.filter === 'to' && action.msg.toAddressHash !== state.addressHash) ||
+        (state.filter === 'from' && action.msg.fromAddressHash !== state.addressHash)) {
+        return state
+      }
+
+      return Object.assign({}, state, { items: [action.msg.tokenTransferHtml, ...state.items] })
+    }
+    case 'RECEIVED_NEW_REWARD': {
+      if (state.channelDisconnected) return state
+
+      return Object.assign({}, state, { items: [action.msg.rewardHtml, ...state.items] })
+    }
+    default:
+      return state
+  }
+}
+
+const elements = {
+  '[data-selector="channel-disconnected-message"]': {
+    render ($el, state) {
+      if (state.channelDisconnected) $el.show()
+    }
+  }
+}
+
+if ($('[data-page="address-token-transfers"]').length) {
+  const store = createAsyncLoadStore(reducer, initialState, 'dataset.identifierHash')
+  const addressHash = $('[data-page="address-details"]')[0].dataset.pageAddressHash
+  const { filter, blockNumber } = humps.camelizeKeys(URI(window.location).query(true))
+
+  connectElements({ store, elements })
+
+  store.dispatch({
+    type: 'PAGE_LOAD',
+    addressHash,
+    filter,
+    beyondPageOne: !!blockNumber
+  })
+
+  const addressChannel = subscribeChannel(`addresses:${addressHash}`)
+  addressChannel.onError(() => store.dispatch({ type: 'CHANNEL_DISCONNECTED' }))
+  addressChannel.on('token_transfer', (msg) => {
+    store.dispatch({
+      type: 'RECEIVED_NEW_TOKEN_TRANSFER',
+      msg: humps.camelizeKeys(msg)
+    })
+  })
+
+  const rewardsChannel = subscribeChannel(`rewards:${addressHash}`)
+  rewardsChannel.onError(() => store.dispatch({ type: 'CHANNEL_DISCONNECTED' }))
+  rewardsChannel.on('new_reward', (msg) => {
+    store.dispatch({
+      type: 'RECEIVED_NEW_REWARD',
+      msg: humps.camelizeKeys(msg)
+    })
+  })
+}

apps/block_scout_web/assets/js/pages/tokens.js

@@ -0,0 +1,80 @@
+import $ from 'jquery'
+import omit from 'lodash/omit'
+import humps from 'humps'
+import socket from '../socket'
+import { createAsyncLoadStore } from '../lib/async_listing_load'
+import { batchChannel } from '../lib/utils'
+import '../app'
+
+const BATCH_THRESHOLD = 10
+
+export const initialState = {
+  channelDisconnected: false,
+  tokenTransferCount: null,
+  tokenTransfersBatch: []
+}
+
+export function reducer (state = initialState, action) {
+  switch (action.type) {
+    case 'ELEMENTS_LOAD': {
+      return Object.assign({}, state, omit(action, 'type'))
+    }
+    case 'CHANNEL_DISCONNECTED': {
+      return Object.assign({}, state, {
+        channelDisconnected: true,
+        tokenTransfersBatch: []
+      })
+    }
+    case 'RECEIVED_NEW_TOKEN_TRANSFER_BATCH': {
+      if (state.channelDisconnected) return state
+
+      const tokenTransferCount = state.tokenTransferCount + action.msgs.length
+
+      const tokenTransfersLength = state.items.length + action.msgs.length
+      if (tokenTransfersLength < BATCH_THRESHOLD) {
+        return Object.assign({}, state, {
+          items: [
+            ...action.msgs.map(msg => msg.tokenTransferHtml).reverse(),
+            ...state.items
+          ],
+          tokenTransferCount
+        })
+      } else if (!state.tokenTransfersBatch.length && action.msgs.length < BATCH_THRESHOLD) {
+        return Object.assign({}, state, {
+          items: [
+            ...action.msgs.map(msg => msg.tokenTransferHtml).reverse(),
+            ...state.items.slice(0, -1 * action.msgs.length)
+          ],
+          tokenTransferCount
+        })
+      } else {
+        return Object.assign({}, state, {
+          tokenTransfersBatch: [
+            ...action.msgs.reverse(),
+            ...state.tokenTransfersBatch
+          ],
+          tokenTransferCount
+        })
+      }
+    }
+    default:
+      return state
+  }
+}
+
+const $tokenTransferListPage = $('[data-page="token-transfer-list"]')
+if ($tokenTransferListPage.length) {
+  const store = createAsyncLoadStore(reducer, initialState, 'dataset.identifierHash')
+
+  const tokenTransfersChannel = socket.channel('token_transfers:new_token_transfer')
+  tokenTransfersChannel.join()
+  tokenTransfersChannel.onError(() => store.dispatch({
+    type: 'CHANNEL_DISCONNECTED'
+  }))
+  tokenTransfersChannel.on('token_transfer', batchChannel((msgs) => {
+    store.dispatch({
+      type: 'RECEIVED_NEW_TOKEN_TRANSFER_BATCH',
+      msgs: humps.camelizeKeys(msgs)
+    })
+  }))
+}

apps/block_scout_web/assets/webpack.config.js

@@ -78,6 +78,7 @@ const appJs =
       'blocks': './js/pages/blocks.js',
       'address': './js/pages/address.js',
       'address-transactions': './js/pages/address/transactions.js',
+      'address-token-transfers': './js/pages/address/token_transfers.js',
       'address-coin-balances': './js/pages/address/coin_balances.js',
       'address-internal-transactions': './js/pages/address/internal_transactions.js',
       'address-logs': './js/pages/address/logs.js',
@@ -87,6 +88,7 @@ const appJs =
       'transaction': './js/pages/transaction.js',
       'verification-form': './js/pages/verification_form.js',
       'token-counters': './js/pages/token_counters.js',
+      'tokens': './js/pages/tokens.js',
       'admin-tasks': './js/pages/admin/tasks.js',
       'read-token-contract': './js/pages/read_token_contract.js',
       'smart-contract-helpers': './js/lib/smart_contract/index.js',

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

@@ -4,14 +4,28 @@ defmodule BlockScoutWeb.AddressChannel do
   """
   use BlockScoutWeb, :channel
 
-  alias BlockScoutWeb.{AddressCoinBalanceView, AddressView, InternalTransactionView, TransactionView}
+  alias BlockScoutWeb.{
-  alias Explorer.{Chain, Market}
+    AddressCoinBalanceView,
-  alias Explorer.Chain.Hash
+    AddressView,
+    InternalTransactionView,
+    TransactionView
+  }
+
+  alias Explorer.{Chain, Market, Repo}
+  alias Explorer.Chain.{Hash, Transaction}
   alias Explorer.Chain.Hash.Address, as: AddressHash
   alias Explorer.ExchangeRates.Token
   alias Phoenix.View
 
-  intercept(["balance_update", "coin_balance", "count", "internal_transaction", "transaction", "verification_result"])
+  intercept([
+    "balance_update",
+    "coin_balance",
+    "count",
+    "internal_transaction",
+    "transaction",
+    "verification_result",
+    "token_transfer"
+  ])
 
   {:ok, burn_address_hash} = Chain.string_to_address_hash("0x0000000000000000000000000000000000000000")
   @burn_address_hash burn_address_hash
@@ -103,6 +117,8 @@ defmodule BlockScoutWeb.AddressChannel do
 
   def handle_out("transaction", data, socket), do: handle_transaction(data, socket, "transaction")
 
+  def handle_out("token_transfer", data, socket), do: handle_token_transfer(data, socket, "token_transfer")
+
   def handle_out("coin_balance", %{block_number: block_number}, socket) do
     coin_balance = Chain.get_coin_balance(socket.assigns.address_hash, block_number)
 
@@ -146,6 +162,34 @@ defmodule BlockScoutWeb.AddressChannel do
     {:noreply, socket}
   end
 
+  def handle_token_transfer(%{address: address, token_transfer: token_transfer}, socket, event) do
+    Gettext.put_locale(BlockScoutWeb.Gettext, socket.assigns.locale)
+
+    transaction =
+      Transaction
+      |> Repo.get_by(hash: token_transfer.transaction_hash)
+      |> Repo.preload([:from_address, :to_address, :block, token_transfers: [:from_address, :to_address, :token]])
+
+    rendered =
+      View.render_to_string(
+        TransactionView,
+        "_tile.html",
+        current_address: address,
+        transaction: transaction,
+        burn_address_hash: @burn_address_hash,
+        conn: socket
+      )
+
+    push(socket, event, %{
+      to_address_hash: to_string(token_transfer.to_address_hash),
+      from_address_hash: to_string(token_transfer.from_address_hash),
+      token_transfer_hash: Hash.to_string(token_transfer.transaction_hash),
+      token_transfer_html: rendered
+    })
+
+    {:noreply, socket}
+  end
+
   defp render_balance_card(address, exchange_rate, locale) do
     Gettext.put_locale(BlockScoutWeb.Gettext, locale)
 

apps/block_scout_web/lib/block_scout_web/channels/token_transfer_channel.ex

@@ -0,0 +1,45 @@
+defmodule BlockScoutWeb.TokenTransferChannel do
+  @moduledoc """
+  Establishes pub/sub channel for live updates of token transfer events.
+  """
+  use BlockScoutWeb, :channel
+
+  alias BlockScoutWeb.Tokens.TransferView
+  alias Explorer.Chain
+  alias Explorer.Chain.Hash
+  alias Phoenix.View
+
+  intercept(["token_transfer"])
+
+  {:ok, burn_address_hash} = Chain.string_to_address_hash("0x0000000000000000000000000000000000000000")
+  @burn_address_hash burn_address_hash
+
+  def join("token_transfers:new_token_transfer", _params, socket) do
+    {:ok, %{}, socket}
+  end
+
+  def join("token_transfers:" <> _transaction_hash, _params, socket) do
+    {:ok, %{}, socket}
+  end
+
+  def handle_out("token_transfer", %{token_transfer: token_transfer}, socket) do
+    Gettext.put_locale(BlockScoutWeb.Gettext, socket.assigns.locale)
+
+    rendered_token_transfer =
+      View.render_to_string(
+        TransferView,
+        "_token_transfer.html",
+        conn: socket,
+        token: token_transfer.token,
+        token_transfer: token_transfer,
+        burn_address_hash: @burn_address_hash
+      )
+
+    push(socket, "token_transfer", %{
+      token_transfer_hash: Hash.to_string(token_transfer.transaction_hash),
+      token_transfer_html: rendered_token_transfer
+    })
+
+    {:noreply, socket}
+  end
+end

apps/block_scout_web/lib/block_scout_web/channels/user_socket.ex

@@ -7,6 +7,7 @@ defmodule BlockScoutWeb.UserSocket do
   channel("exchange_rate:*", BlockScoutWeb.ExchangeRateChannel)
   channel("rewards:*", BlockScoutWeb.RewardChannel)
   channel("transactions:*", BlockScoutWeb.TransactionChannel)
+  channel("token_transfers:*", BlockScoutWeb.TokenTransferChannel)
 
   def connect(%{"locale" => locale}, socket) do
     {:ok, assign(socket, :locale, locale)}

apps/block_scout_web/lib/block_scout_web/notifier.ex

@@ -6,7 +6,7 @@ defmodule BlockScoutWeb.Notifier do
   alias Absinthe.Subscription
   alias BlockScoutWeb.{AddressContractVerificationView, Endpoint}
   alias Explorer.{Chain, Market, Repo}
-  alias Explorer.Chain.{Address, InternalTransaction, Transaction}
+  alias Explorer.Chain.{Address, InternalTransaction, TokenTransfer, Transaction}
   alias Explorer.Chain.Supply.RSK
   alias Explorer.Chain.Transaction.History.TransactionStats
   alias Explorer.Counters.AverageBlockTime
@@ -112,6 +112,20 @@ defmodule BlockScoutWeb.Notifier do
         token_transfers,
         token_transfers: token_contract_address_hash
       )
+
+      token_transfers_full =
+        token_transfers
+        |> Stream.map(
+          &(TokenTransfer
+            |> Repo.get_by(
+              transaction_hash: &1.transaction_hash,
+              token_contract_address_hash: &1.token_contract_address_hash
+            )
+            |> Repo.preload([:from_address, :to_address, :token, transaction: :block]))
+        )
+
+      token_transfers_full
+      |> Enum.each(&broadcast_token_transfer/1)
     end
   end
 
@@ -123,10 +137,13 @@ defmodule BlockScoutWeb.Notifier do
         :block => :optional,
         [created_contract_address: :names] => :optional,
         [from_address: :names] => :optional,
-        [to_address: :names] => :optional,
+        [to_address: :names] => :optional
-        :token_transfers => :optional
       }
     )
+    |> Enum.map(fn tx ->
+      # Disable parsing of token transfers from websocket for transaction tab because we display token transfers at a separate tab
+      Map.put(tx, :token_transfers, [])
+    end)
     |> Enum.each(&broadcast_transaction/1)
   end
 
@@ -248,4 +265,28 @@ defmodule BlockScoutWeb.Notifier do
       })
     end
   end
+
+  defp broadcast_token_transfer(token_transfer) do
+    broadcast_token_transfer(token_transfer, "token_transfers:new_token_transfer", "token_transfer")
+  end
+
+  defp broadcast_token_transfer(token_transfer, token_transfer_channel, event) do
+    Endpoint.broadcast("token_transfers:#{token_transfer.transaction_hash}", event, %{})
+
+    Endpoint.broadcast(token_transfer_channel, event, %{
+      token_transfer: token_transfer
+    })
+
+    Endpoint.broadcast("addresses:#{token_transfer.from_address_hash}", event, %{
+      address: token_transfer.from_address,
+      token_transfer: token_transfer
+    })
+
+    if token_transfer.to_address_hash != token_transfer.from_address_hash do
+      Endpoint.broadcast("addresses:#{token_transfer.to_address_hash}", event, %{
+        address: token_transfer.to_address,
+        token_transfer: token_transfer
+      })
+    end
+  end
 end

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

@@ -1,7 +1,7 @@
 <section class="container">
   <%= render BlockScoutWeb.AddressView, "overview.html", assigns %>
 
-  <section>
+  <section data-page="address-token-transfers">
     <div class="card">
       <%= render BlockScoutWeb.AddressView, "_tabs.html", assigns %>
       <div data-async-load data-async-listing="<%= @current_path %>" class="card-body">
@@ -74,6 +74,6 @@
 
       </div>
     </div>
-    <script defer data-cfasync="false" src="<%= static_path(@conn, "/js/address.js") %>"></script>
+    <script defer data-cfasync="false" src="<%= static_path(@conn, "/js/address-token-transfers.js") %>"></script>
   </section>
 </section>

apps/block_scout_web/lib/block_scout_web/templates/tokens/transfer/index.html.eex

@@ -7,7 +7,7 @@
     conn: @conn
   ) %>
 
-  <section>
+  <section data-page="token-transfer-list">
     <div class="card">
       <%= render OverviewView, "_tabs.html", assigns %>
       <div class="card-body" data-async-load data-async-listing="<%= @current_path %>">
@@ -35,4 +35,5 @@
     </div>
   </section>
   <script defer data-cfasync="false" src="<%= static_path(@conn, "/js/token-counters.js") %>"></script>
+  <script defer data-cfasync="false" src="<%= static_path(@conn, "/js/tokens.js") %>"></script>
 </section>