Fix zero balances coming via WS (#9510)

CHANGELOG.md

@@ -37,6 +37,7 @@
 - [#9529](https://github.com/blockscout/blockscout/pull/9529) - Fix `MAX_SAFE_INTEGER` frontend bug
 - [#9518](https://github.com/blockscout/blockscout/pull/9518), [#9628](https://github.com/blockscout/blockscout/pull/9628) - Fix MultipleResultsError in `smart_contract_creation_tx_bytecode/1`
 - [#9514](https://github.com/blockscout/blockscout/pull/9514) - Fix missing `0x` prefix for `blockNumber`, `logIndex`, `transactionIndex` and remove `transactionLogIndex` in `eth_getLogs` response.
+- [#9510](https://github.com/blockscout/blockscout/pull/9510) - Fix WS false 0 token balances
 - [#9512](https://github.com/blockscout/blockscout/pull/9512) - Docker-compose 2.24.6 compatibility
 - [#9262](https://github.com/blockscout/blockscout/pull/9262) - Fix withdrawal status
 - [#9123](https://github.com/blockscout/blockscout/pull/9123) - Fixes in Optimism due to changed log topics type

apps/block_scout_web/test/block_scout_web/controllers/api/v2/address_controller_test.exs

@@ -1,7 +1,9 @@
 defmodule BlockScoutWeb.API.V2.AddressControllerTest do
   use BlockScoutWeb.ConnCase
   use EthereumJSONRPC.Case, async: false
+  use BlockScoutWeb.ChannelCase
 
+  alias ABI.{TypeDecoder, TypeEncoder}
   alias BlockScoutWeb.Models.UserFromAuth
   alias Explorer.{Chain, Repo}
   alias Explorer.Chain.Address.Counters
@@ -1835,7 +1837,7 @@ defmodule BlockScoutWeb.API.V2.AddressControllerTest do
           )
           |> Repo.preload([:token])
         end
-        |> Enum.sort_by(fn x -> x.value end, :asc)
+        |> Enum.sort_by(fn x -> Decimal.to_integer(x.value) end, :asc)
 
       ctbs_erc_1155 =
         for _ <- 0..50 do
@@ -1846,7 +1848,7 @@ defmodule BlockScoutWeb.API.V2.AddressControllerTest do
           )
           |> Repo.preload([:token])
         end
-        |> Enum.sort_by(fn x -> x.value end, :asc)
+        |> Enum.sort_by(fn x -> Decimal.to_integer(x.value) end, :asc)
 
       filter = %{"type" => "ERC-20"}
       request = get(conn, "/api/v2/addresses/#{address.hash}/tokens", filter)
@@ -1883,6 +1885,291 @@ defmodule BlockScoutWeb.API.V2.AddressControllerTest do
     end
   end
 
+  describe "checks TokenBalanceOnDemand" do
+    setup do
+      Supervisor.terminate_child(Explorer.Supervisor, Explorer.Chain.Cache.BlockNumber.child_id())
+      Supervisor.restart_child(Explorer.Supervisor, Explorer.Chain.Cache.BlockNumber.child_id())
+      old_env = Application.get_env(:indexer, Indexer.Fetcher.TokenBalanceOnDemand)
+
+      Application.put_env(
+        :indexer,
+        Indexer.Fetcher.TokenBalanceOnDemand,
+        Keyword.put(old_env, :fallback_threshold_in_blocks, 0)
+      )
+
+      on_exit(fn ->
+        Application.put_env(:indexer, Indexer.Fetcher.TokenBalanceOnDemand, old_env)
+      end)
+    end
+
+    test "Indexer.Fetcher.TokenBalanceOnDemand broadcasts only updated balances", %{conn: conn} do
+      address = insert(:address)
+
+      ctbs_erc_20 =
+        for i <- 0..1 do
+          ctb =
+            insert(:address_current_token_balance_with_token_id_and_fixed_token_type,
+              address: address,
+              token_type: "ERC-20",
+              token_id: nil
+            )
+
+          {to_string(ctb.token_contract_address_hash),
+           Decimal.to_integer(ctb.value) + if(rem(i, 2) == 0, do: 1, else: 0)}
+        end
+        |> Enum.into(%{})
+
+      ctbs_erc_721 =
+        for i <- 0..1 do
+          ctb =
+            insert(:address_current_token_balance_with_token_id_and_fixed_token_type,
+              address: address,
+              token_type: "ERC-721",
+              token_id: nil
+            )
+
+          {to_string(ctb.token_contract_address_hash),
+           Decimal.to_integer(ctb.value) + if(rem(i, 2) == 0, do: 1, else: 0)}
+        end
+        |> Enum.into(%{})
+
+      other_balances = Map.merge(ctbs_erc_20, ctbs_erc_721)
+
+      balances_erc_1155 =
+        for i <- 0..1 do
+          ctb =
+            insert(:address_current_token_balance_with_token_id_and_fixed_token_type,
+              address: address,
+              token_type: "ERC-1155",
+              token_id: Enum.random(1..100_000)
+            )
+
+          {{to_string(ctb.token_contract_address_hash), to_string(ctb.token_id)},
+           Decimal.to_integer(ctb.value) + if(rem(i, 2) == 0, do: 1, else: 0)}
+        end
+        |> Enum.into(%{})
+
+      block_number_hex = "0x" <> (Integer.to_string(insert(:block).number, 16) |> String.upcase())
+
+      expect(EthereumJSONRPC.Mox, :json_rpc, fn [
+                                                  %{
+                                                    id: id_1,
+                                                    jsonrpc: "2.0",
+                                                    method: "eth_call",
+                                                    params: [
+                                                      %{
+                                                        data: "0x00fdd58e" <> request_1,
+                                                        to: contract_address_1
+                                                      },
+                                                      ^block_number_hex
+                                                    ]
+                                                  },
+                                                  %{
+                                                    id: id_2,
+                                                    jsonrpc: "2.0",
+                                                    method: "eth_call",
+                                                    params: [
+                                                      %{
+                                                        data: "0x00fdd58e" <> request_2,
+                                                        to: contract_address_2
+                                                      },
+                                                      ^block_number_hex
+                                                    ]
+                                                  }
+                                                ],
+                                                _options ->
+        types_list = [:address, {:uint, 256}]
+
+        [address_1, token_id_1] = request_1 |> Base.decode16!(case: :lower) |> TypeDecoder.decode_raw(types_list)
+
+        assert address_1 == address.hash.bytes
+
+        result_1 =
+          balances_erc_1155[{contract_address_1 |> String.downcase(), to_string(token_id_1)}]
+          |> List.wrap()
+          |> TypeEncoder.encode_raw([{:uint, 256}], :standard)
+          |> Base.encode16(case: :lower)
+
+        [address_2, token_id_2] = request_2 |> Base.decode16!(case: :lower) |> TypeDecoder.decode_raw(types_list)
+
+        assert address_2 == address.hash.bytes
+
+        result_2 =
+          balances_erc_1155[{contract_address_2 |> String.downcase(), to_string(token_id_2)}]
+          |> List.wrap()
+          |> TypeEncoder.encode_raw([{:uint, 256}], :standard)
+          |> Base.encode16(case: :lower)
+
+        {:ok,
+         [
+           %{
+             id: id_1,
+             jsonrpc: "2.0",
+             result: "0x" <> result_1
+           },
+           %{
+             id: id_2,
+             jsonrpc: "2.0",
+             result: "0x" <> result_2
+           }
+         ]}
+      end)
+
+      expect(EthereumJSONRPC.Mox, :json_rpc, fn [
+                                                  %{
+                                                    id: id_1,
+                                                    jsonrpc: "2.0",
+                                                    method: "eth_call",
+                                                    params: [
+                                                      %{
+                                                        data: "0x70a08231" <> request_1,
+                                                        to: contract_address_1
+                                                      },
+                                                      ^block_number_hex
+                                                    ]
+                                                  },
+                                                  %{
+                                                    id: id_2,
+                                                    jsonrpc: "2.0",
+                                                    method: "eth_call",
+                                                    params: [
+                                                      %{
+                                                        data: "0x70a08231" <> request_2,
+                                                        to: contract_address_2
+                                                      },
+                                                      ^block_number_hex
+                                                    ]
+                                                  },
+                                                  %{
+                                                    id: id_3,
+                                                    jsonrpc: "2.0",
+                                                    method: "eth_call",
+                                                    params: [
+                                                      %{
+                                                        data: "0x70a08231" <> request_3,
+                                                        to: contract_address_3
+                                                      },
+                                                      ^block_number_hex
+                                                    ]
+                                                  },
+                                                  %{
+                                                    id: id_4,
+                                                    jsonrpc: "2.0",
+                                                    method: "eth_call",
+                                                    params: [
+                                                      %{
+                                                        data: "0x70a08231" <> request_4,
+                                                        to: contract_address_4
+                                                      },
+                                                      ^block_number_hex
+                                                    ]
+                                                  }
+                                                ],
+                                                _options ->
+        types_list = [:address]
+
+        assert request_1 |> Base.decode16!(case: :lower) |> TypeDecoder.decode_raw(types_list) == [address.hash.bytes]
+
+        assert request_2 |> Base.decode16!(case: :lower) |> TypeDecoder.decode_raw(types_list) == [address.hash.bytes]
+
+        assert request_3 |> Base.decode16!(case: :lower) |> TypeDecoder.decode_raw(types_list) == [address.hash.bytes]
+
+        assert request_4 |> Base.decode16!(case: :lower) |> TypeDecoder.decode_raw(types_list) == [address.hash.bytes]
+
+        result_1 =
+          other_balances[contract_address_1 |> String.downcase()]
+          |> List.wrap()
+          |> TypeEncoder.encode_raw([{:uint, 256}], :standard)
+          |> Base.encode16(case: :lower)
+
+        result_2 =
+          other_balances[contract_address_2 |> String.downcase()]
+          |> List.wrap()
+          |> TypeEncoder.encode_raw([{:uint, 256}], :standard)
+          |> Base.encode16(case: :lower)
+
+        result_3 =
+          other_balances[contract_address_3 |> String.downcase()]
+          |> List.wrap()
+          |> TypeEncoder.encode_raw([{:uint, 256}], :standard)
+          |> Base.encode16(case: :lower)
+
+        result_4 =
+          other_balances[contract_address_4 |> String.downcase()]
+          |> List.wrap()
+          |> TypeEncoder.encode_raw([{:uint, 256}], :standard)
+          |> Base.encode16(case: :lower)
+
+        {:ok,
+         [
+           %{
+             id: id_1,
+             jsonrpc: "2.0",
+             result: "0x" <> result_1
+           },
+           %{
+             id: id_2,
+             jsonrpc: "2.0",
+             result: "0x" <> result_2
+           },
+           %{
+             id: id_3,
+             jsonrpc: "2.0",
+             result: "0x" <> result_3
+           },
+           %{
+             id: id_4,
+             jsonrpc: "2.0",
+             result: "0x" <> result_4
+           }
+         ]}
+      end)
+
+      topic = "addresses:#{address.hash}"
+
+      {:ok, _reply, _socket} =
+        BlockScoutWeb.UserSocketV2
+        |> socket("no_id", %{})
+        |> subscribe_and_join(topic)
+
+      request = get(conn, "/api/v2/addresses/#{address.hash}/tokens")
+      assert _response = json_response(request, 200)
+      overflow = false
+
+      assert_receive %Phoenix.Socket.Message{
+                       payload: %{token_balances: [ctb_erc_20], overflow: ^overflow},
+                       event: "updated_token_balances_erc_20",
+                       topic: ^topic
+                     },
+                     :timer.seconds(1)
+
+      assert_receive %Phoenix.Socket.Message{
+                       payload: %{token_balances: [ctb_erc_721], overflow: ^overflow},
+                       event: "updated_token_balances_erc_721",
+                       topic: ^topic
+                     },
+                     :timer.seconds(1)
+
+      assert_receive %Phoenix.Socket.Message{
+                       payload: %{token_balances: [ctb_erc_1155], overflow: ^overflow},
+                       event: "updated_token_balances_erc_1155",
+                       topic: ^topic
+                     },
+                     :timer.seconds(1)
+
+      assert Decimal.to_integer(ctb_erc_20["value"]) ==
+               other_balances[ctb_erc_20["token"]["address"] |> String.downcase()]
+
+      assert Decimal.to_integer(ctb_erc_721["value"]) ==
+               other_balances[ctb_erc_721["token"]["address"] |> String.downcase()]
+
+      assert Decimal.to_integer(ctb_erc_1155["value"]) ==
+               balances_erc_1155[
+                 {ctb_erc_1155["token"]["address"] |> String.downcase(), to_string(ctb_erc_1155["token_id"])}
+               ]
+    end
+  end
+
   describe "/addresses/{address_hash}/withdrawals" do
     test "get empty list on non existing address", %{conn: conn} do
       address = build(:address)

apps/explorer/config/test.exs

@@ -86,4 +86,3 @@ config :explorer, Explorer.ExchangeRates.Source.TransactionAndLog,
 
 config :explorer, Explorer.Chain.Fetcher.CheckBytecodeMatchingOnDemand, enabled: false
 config :explorer, Explorer.Chain.Fetcher.FetchValidatorInfoOnDemand, enabled: false
-config :explorer, Explorer.Tags.AddressTag.Cataloger, enabled: false

apps/explorer/lib/explorer/chain/address/current_token_balance.ex

@@ -160,9 +160,7 @@ defmodule Explorer.Chain.Address.CurrentTokenBalance do
       on: ctb.token_contract_address_hash == t.contract_address_hash,
       preload: [token: t],
       select: ctb,
-      select_merge: ^%{fiat_value: fiat_balance},
-      order_by: ^[desc_nulls_last: fiat_balance],
-      order_by: [desc: ctb.value, desc: ctb.id]
+      select_merge: ^%{fiat_value: fiat_balance}
     )
   end
 

apps/indexer/lib/indexer/fetcher/token_balance_on_demand.ex

@@ -71,9 +71,16 @@ defmodule Indexer.Fetcher.TokenBalanceOnDemand do
   end
 
   defp fetch_and_update(block_number, address_hash, stale_current_token_balances) do
-    %{erc_1155: erc_1155_ctbs, other: other_ctbs, tokens: tokens} =
+    %{
+      erc_1155: erc_1155_ctbs,
+      other: other_ctbs,
+      tokens: tokens,
+      balances_map: balances_map
+    } =
       stale_current_token_balances
-      |> Enum.reduce(%{erc_1155: [], other: [], tokens: %{}}, fn %{token_id: token_id} = stale_current_token_balance,
+      |> Enum.reduce(%{erc_1155: [], other: [], tokens: %{}, balances_map: %{}}, fn %{
+                                                                                      token_id: token_id
+                                                                                    } = stale_current_token_balance,
                                                                                     acc ->
         prepared_ctb = %{
           token_contract_address_hash:
@@ -98,35 +105,40 @@ defmodule Indexer.Fetcher.TokenBalanceOnDemand do
             Map.put(acc, :other, [prepared_ctb | acc[:other]])
           end
 
-        Map.put(result, :tokens, updated_tokens)
-      end)
+        updated_balances_map =
+          Map.put(
+            acc[:balances_map],
+            ctb_to_key(stale_current_token_balance),
+            stale_current_token_balance.value
+          )
 
-    erc_1155_ctbs_reversed = Enum.reverse(erc_1155_ctbs)
-    other_ctbs_reversed = Enum.reverse(other_ctbs)
+        result
+        |> Map.put(:tokens, updated_tokens)
+        |> Map.put(:balances_map, updated_balances_map)
+      end)
 
     updated_erc_1155_ctbs =
-      if Enum.count(erc_1155_ctbs_reversed) > 0 do
-        erc_1155_ctbs_reversed
+      if Enum.count(erc_1155_ctbs) > 0 do
+        erc_1155_ctbs
         |> BalanceReader.get_balances_of_erc_1155()
-        |> Enum.zip(erc_1155_ctbs_reversed)
+        |> Enum.zip(erc_1155_ctbs)
         |> Enum.map(&prepare_updated_balance(&1, block_number))
       else
         []
       end
 
     updated_other_ctbs =
-      if Enum.count(other_ctbs_reversed) > 0 do
-        other_ctbs_reversed
+      if Enum.count(other_ctbs) > 0 do
+        other_ctbs
         |> BalanceReader.get_balances_of()
-        |> Enum.zip(other_ctbs_reversed)
+        |> Enum.zip(other_ctbs)
         |> Enum.map(&prepare_updated_balance(&1, block_number))
       else
         []
       end
 
     filtered_current_token_balances_update_params =
-      (updated_erc_1155_ctbs ++ updated_other_ctbs)
-      |> Enum.filter(&(!is_nil(&1)))
+      (updated_erc_1155_ctbs ++ updated_other_ctbs) |> Enum.filter(&(!is_nil(&1)))
 
     if Enum.count(filtered_current_token_balances_update_params) > 0 do
       {:ok,
@@ -140,12 +152,14 @@ defmodule Indexer.Fetcher.TokenBalanceOnDemand do
           broadcast: false
         })
 
+      filtered_imported_ctbs = filter_imported_ctbs(imported_ctbs, balances_map)
+
       Publisher.broadcast(
         %{
           address_current_token_balances: %{
             address_hash: to_string(address_hash),
             address_current_token_balances:
-              imported_ctbs
+              filtered_imported_ctbs
               |> Enum.map(fn ctb -> %CurrentTokenBalance{ctb | token: tokens[ctb.token_contract_address_hash.bytes]} end)
           }
         },
@@ -154,6 +168,21 @@ defmodule Indexer.Fetcher.TokenBalanceOnDemand do
     end
   end
 
+  defp filter_imported_ctbs(imported_ctbs, balances_map) do
+    Enum.filter(imported_ctbs, fn ctb ->
+      if balance = balances_map[ctb_to_key(ctb)] do
+        Decimal.compare(balance, ctb.value) != :eq
+      else
+        Logger.error("Imported unknown balance")
+        true
+      end
+    end)
+  end
+
+  defp ctb_to_key(ctb) do
+    {ctb.token_contract_address_hash.bytes, ctb.token_type, ctb.token_id && Decimal.to_integer(ctb.token_id)}
+  end
+
   defp prepare_updated_balance({{:ok, updated_balance}, stale_current_token_balance}, block_number) do
     %{}
     |> Map.put(:address_hash, stale_current_token_balance.address_hash)