fix: Fix empty current token balances (#10745)

apps/explorer/config/config.exs

@@ -123,6 +123,7 @@ config :explorer, Explorer.Migrator.SanitizeIncorrectWETHTokenTransfers, enabled
 config :explorer, Explorer.Migrator.TransactionBlockConsensus, enabled: true
 config :explorer, Explorer.Migrator.TokenTransferBlockConsensus, enabled: true
 config :explorer, Explorer.Migrator.RestoreOmittedWETHTransfers, enabled: true
+config :explorer, Explorer.Migrator.SanitizeMissingTokenBalances, enabled: true
 
 config :explorer, Explorer.Chain.Fetcher.CheckBytecodeMatchingOnDemand, enabled: true
 

apps/explorer/config/runtime/test.exs

@@ -49,6 +49,7 @@ config :explorer, Explorer.Migrator.TransactionBlockConsensus, enabled: false
 config :explorer, Explorer.Migrator.TokenTransferBlockConsensus, enabled: false
 config :explorer, Explorer.Migrator.ShrinkInternalTransactions, enabled: false
 config :explorer, Explorer.Migrator.RestoreOmittedWETHTransfers, enabled: false
+config :explorer, Explorer.Migrator.SanitizeMissingTokenBalances, enabled: false
 
 config :explorer,
   realtime_events_sender: Explorer.Chain.Events.SimpleSender

apps/explorer/lib/explorer/application.ex

@@ -144,7 +144,8 @@ defmodule Explorer.Application do
         configure(Explorer.Migrator.RestoreOmittedWETHTransfers),
         configure(Explorer.Migrator.FilecoinPendingAddressOperations),
         configure_mode_dependent_process(Explorer.Migrator.ShrinkInternalTransactions, :indexer),
-        configure_chain_type_dependent_process(Explorer.Chain.Cache.StabilityValidatorsCounters, :stability)
+        configure_chain_type_dependent_process(Explorer.Chain.Cache.StabilityValidatorsCounters, :stability),
+        configure_mode_dependent_process(Explorer.Migrator.SanitizeMissingTokenBalances, :indexer)
       ]
       |> List.flatten()
 

apps/explorer/lib/explorer/chain/import/runner/address/current_token_balances.ex

@@ -108,17 +108,17 @@ defmodule Explorer.Chain.Import.Runner.Address.CurrentTokenBalances do
       |> Map.put(:timestamps, timestamps)
 
     multi
-    |> Multi.run(:filter_placeholders, fn _, _ ->
+    |> Multi.run(:filter_ctb_placeholders, fn _, _ ->
       Instrumenter.block_import_stage_runner(
         fn -> TokenBalances.filter_placeholders(changes_list) end,
         :block_following,
         :current_token_balances,
-        :filter_placeholders
+        :filter_ctb_placeholders
       )
     end)
-    |> Multi.run(:address_current_token_balances, fn repo, _ ->
+    |> Multi.run(:address_current_token_balances, fn repo, %{filter_ctb_placeholders: filtered_changes_list} ->
       Instrumenter.block_import_stage_runner(
-        fn -> insert(repo, changes_list, insert_options) end,
+        fn -> insert(repo, filtered_changes_list, insert_options) end,
         :block_following,
         :current_token_balances,
         :address_current_token_balances

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

@@ -13,6 +13,7 @@ defmodule Explorer.Chain.Import.Stage.BlockFollowing do
     do: [
       Runner.Block.SecondDegreeRelations,
       Runner.Block.Rewards,
+      Runner.Address.TokenBalances,
       Runner.Address.CurrentTokenBalances
     ]
 

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

@@ -13,7 +13,6 @@ defmodule Explorer.Chain.Import.Stage.BlockReferencing do
     Runner.Logs,
     Runner.Tokens,
     Runner.TokenInstances,
-    Runner.Address.TokenBalances,
     Runner.TransactionActions,
     Runner.Withdrawals
   ]

apps/explorer/lib/explorer/migrator/sanitize_missing_token_balances.ex

@@ -0,0 +1,61 @@
+defmodule Explorer.Migrator.SanitizeMissingTokenBalances do
+  @moduledoc """
+  Set value and value_fetched_at to nil for those token balances that are already filled but their
+  current token balances are not so the token balance fetcher could re-fetch them.
+  """
+
+  use Explorer.Migrator.FillingMigration
+
+  import Ecto.Query
+
+  alias Explorer.Chain.Address.{CurrentTokenBalance, TokenBalance}
+  alias Explorer.Migrator.FillingMigration
+  alias Explorer.Repo
+
+  @migration_name "sanitize_missing_token_balances"
+
+  @impl FillingMigration
+  def migration_name, do: @migration_name
+
+  @impl FillingMigration
+  def last_unprocessed_identifiers(state) do
+    limit = batch_size() * concurrency()
+
+    ids =
+      unprocessed_data_query()
+      |> select([_ctb, tb], tb.id)
+      |> limit(^limit)
+      |> Repo.all(timeout: :infinity)
+
+    {ids, state}
+  end
+
+  @impl FillingMigration
+  def unprocessed_data_query do
+    from(
+      ctb in CurrentTokenBalance,
+      join: tb in TokenBalance,
+      on:
+        ctb.address_hash == tb.address_hash and
+          ctb.token_contract_address_hash == tb.token_contract_address_hash and
+          ctb.block_number == tb.block_number and
+          ((is_nil(ctb.token_id) and is_nil(tb.token_id)) or ctb.token_id == tb.token_id),
+      where: is_nil(ctb.value) or is_nil(ctb.value_fetched_at),
+      where: not is_nil(tb.value) and not is_nil(tb.value_fetched_at)
+    )
+  end
+
+  @impl FillingMigration
+  def update_batch(token_balance_ids) do
+    query =
+      from(tb in TokenBalance,
+        where: tb.id in ^token_balance_ids,
+        update: [set: [value: nil, value_fetched_at: nil]]
+      )
+
+    Repo.update_all(query, [], timeout: :infinity)
+  end
+
+  @impl FillingMigration
+  def update_cache, do: :ok
+end

apps/explorer/test/explorer/migrator/sanitize_missing_token_balances_test.exs

@@ -0,0 +1,40 @@
+defmodule Explorer.Migrator.SanitizeMissingTokenBalancesTest do
+  use Explorer.DataCase, async: false
+
+  alias Explorer.Chain.Address.TokenBalance
+  alias Explorer.Migrator.{SanitizeMissingTokenBalances, MigrationStatus}
+  alias Explorer.Repo
+
+  describe "Migrate token balances" do
+    test "Unset value and value_fetched_at for token balances related to not processed current token balances" do
+      Enum.each(0..10, fn _x ->
+        token_balance = insert(:token_balance)
+
+        insert(:address_current_token_balance,
+          address: token_balance.address,
+          token_contract_address_hash: token_balance.token_contract_address_hash,
+          token_id: token_balance.token_id,
+          block_number: token_balance.block_number,
+          value: nil,
+          value_fetched_at: nil
+        )
+
+        refute is_nil(token_balance.value)
+        refute is_nil(token_balance.value_fetched_at)
+      end)
+
+      assert MigrationStatus.get_status("sanitize_missing_token_balances") == nil
+
+      SanitizeMissingTokenBalances.start_link([])
+      Process.sleep(100)
+
+      TokenBalance
+      |> Repo.all()
+      |> Enum.each(fn tb ->
+        assert %{value: nil, value_fetched_at: nil} = tb
+      end)
+
+      assert MigrationStatus.get_status("sanitize_missing_token_balances") == "completed"
+    end
+  end
+end