Merge branch 'master' of https://github.com/blockscout/blockscout int…

…o sync-11-25-24

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

@@ -13,7 +13,7 @@ defmodule Explorer.Chain.Address.CurrentTokenBalance do
   import Explorer.Chain.SmartContract, only: [burn_address_hash_string: 0]
   import Explorer.Chain.SmartContract.Proxy.Models.Implementation, only: [proxy_implementations_association: 0]
 
-  alias Explorer.{Chain, PagingOptions, Repo}
+  alias Explorer.{Chain, Helper, PagingOptions, Repo}
   alias Explorer.Chain.{Address, Block, CurrencyHelper, Hash, Token}
   alias Explorer.Chain.Address.TokenBalance
 
@@ -354,4 +354,25 @@ defmodule Explorer.Chain.Address.CurrentTokenBalance do
 
     Stream.concat([row_names], holders_list)
   end
+
+  @doc """
+  Encode `address_hash`, `token_contract_address_hash` and `token_id` into a string that can be used in
+  `(address_hash, token_contract_address_hash, token_id) IN (...)` WHERE clause
+  """
+  @spec encode_ids([{Hash.t(), Hash.t(), non_neg_integer()}] | [{Hash.t(), Hash.t()}]) :: binary()
+  def encode_ids(ids) do
+    encoded_values =
+      ids
+      |> Enum.reduce("", fn
+        {address_hash, token_hash, token_id}, acc ->
+          acc <>
+            "('#{Helper.hash_to_query_string(address_hash)}', '#{Helper.hash_to_query_string(token_hash)}', #{token_id}),"
+
+        {address_hash, token_hash}, acc ->
+          acc <> "('#{Helper.hash_to_query_string(address_hash)}', '#{Helper.hash_to_query_string(token_hash)}'),"
+      end)
+      |> String.trim_trailing(",")
+
+    "(#{encoded_values})"
+  end
 end

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

@@ -116,7 +116,15 @@ defmodule Explorer.Chain.Import.Runner.Address.CurrentTokenBalances do
         :filter_ctb_placeholders
       )
     end)
-    |> Multi.run(:address_current_token_balances, fn repo, %{filter_ctb_placeholders: filtered_changes_list} ->
+    |> Multi.run(:filter_params, fn repo, %{filter_ctb_placeholders: filtered_changes_list} ->
+      Instrumenter.block_import_stage_runner(
+        fn -> filter_params(repo, filtered_changes_list) end,
+        :block_following,
+        :current_token_balances,
+        :filter_params
+      )
+    end)
+    |> Multi.run(:address_current_token_balances, fn repo, %{filter_params: filtered_changes_list} ->
       Instrumenter.block_import_stage_runner(
         fn -> insert(repo, filtered_changes_list, insert_options) end,
         :block_following,
@@ -205,6 +213,81 @@ defmodule Explorer.Chain.Import.Runner.Address.CurrentTokenBalances do
     end
   end
 
+  defp filter_params(repo, changes_list) do
+    {params_without_token_id, params_with_token_id} = Enum.split_with(changes_list, &is_nil(&1[:token_id]))
+
+    existing_ctb_without_token_id = select_existing_current_token_balances(repo, params_without_token_id, false)
+    existing_ctb_with_token_id = select_existing_current_token_balances(repo, params_with_token_id, true)
+
+    existing_ctb_map =
+      existing_ctb_without_token_id
+      |> Enum.concat(existing_ctb_with_token_id)
+      |> Map.new(fn [address_hash, token_contract_address_hash, token_id, block_number, value, value_fetched_at] ->
+        {{address_hash, token_contract_address_hash, token_id},
+         %{block_number: block_number, value: value, value_fetched_at: value_fetched_at}}
+      end)
+
+    filtered_ctbs =
+      Enum.filter(changes_list, fn ctb ->
+        existing_ctb = existing_ctb_map[{ctb[:address_hash], ctb[:token_contract_address_hash], ctb[:token_id]}]
+        should_update?(ctb, existing_ctb)
+      end)
+
+    {:ok, filtered_ctbs}
+  end
+
+  defp select_existing_current_token_balances(_repo, [], _with_token_id?), do: []
+
+  defp select_existing_current_token_balances(repo, params, with_token_id?) do
+    params
+    |> existing_ctb_query(with_token_id?)
+    |> repo.query!()
+    |> Map.get(:rows)
+  end
+
+  defp existing_ctb_query(params, false) do
+    encoded_ids =
+      params
+      |> Enum.map(&{&1.address_hash, &1.token_contract_address_hash})
+      |> CurrentTokenBalance.encode_ids()
+
+    """
+    SELECT ctb.address_hash, ctb.token_contract_address_hash, ctb.token_id, ctb.block_number, ctb.value, ctb.value_fetched_at
+    FROM address_current_token_balances ctb
+    WHERE (ctb.address_hash, ctb.token_contract_address_hash) IN #{encoded_ids} AND ctb.token_id IS NULL
+    """
+  end
+
+  defp existing_ctb_query(params, true) do
+    encoded_ids =
+      params
+      |> Enum.map(&{&1.address_hash, &1.token_contract_address_hash, &1.token_id})
+      |> CurrentTokenBalance.encode_ids()
+
+    """
+    SELECT ctb.address_hash, ctb.token_contract_address_hash, ctb.token_id, ctb.block_number, ctb.value, ctb.value_fetched_at
+    FROM address_current_token_balances ctb
+    WHERE (ctb.address_hash, ctb.token_contract_address_hash, ctb.token_id) IN #{encoded_ids}
+    """
+  end
+
+  # ctb does not exist
+  defp should_update?(_new_ctb, nil), do: true
+
+  # new ctb has no value
+  defp should_update?(%{value_fetched_at: nil}, _existing_ctb), do: false
+
+  # new ctb is newer
+  defp should_update?(%{block_number: new_ctb_block_number}, %{block_number: existing_ctb_block_number})
+       when new_ctb_block_number > existing_ctb_block_number,
+       do: true
+
+  # new ctb is the same height or older
+  defp should_update?(new_ctb, existing_ctb) do
+    existing_ctb.block_number == new_ctb.block_number and not is_nil(new_ctb.value) and
+      (is_nil(existing_ctb.value_fetched_at) or existing_ctb.value_fetched_at < new_ctb.value_fetched_at)
+  end
+
   @spec insert(Repo.t(), [map()], %{
           optional(:on_conflict) => Import.Runner.on_conflict(),
           required(:timeout) => timeout(),

apps/explorer/lib/explorer/chain/token_transfer.ex

@@ -137,7 +137,7 @@ defmodule Explorer.Chain.TokenTransfer do
 
   import Ecto.Changeset
 
-  alias Explorer.Chain
+  alias Explorer.{Chain, Helper}
   alias Explorer.Chain.{DenormalizationHelper, Hash, Log, TokenTransfer}
   alias Explorer.Chain.SmartContract.Proxy.Models.Implementation
   alias Explorer.{PagingOptions, Repo}
@@ -648,22 +648,13 @@ defmodule Explorer.Chain.TokenTransfer do
     encoded_values =
       ids
       |> Enum.reduce("", fn {t_hash, b_hash, log_index}, acc ->
-        acc <> "('#{hash_to_query_string(t_hash)}', '#{hash_to_query_string(b_hash)}', #{log_index}),"
+        acc <> "('#{Helper.hash_to_query_string(t_hash)}', '#{Helper.hash_to_query_string(b_hash)}', #{log_index}),"
       end)
       |> String.trim_trailing(",")
 
     "(#{encoded_values})"
   end
 
-  defp hash_to_query_string(hash) do
-    s_hash =
-      hash
-      |> to_string()
-      |> String.trim_leading("0")
-
-    "\\#{s_hash}"
-  end
-
   @doc """
   Fetches token transfers from logs.
   """

apps/explorer/lib/explorer/helper.ex

@@ -5,7 +5,7 @@ defmodule Explorer.Helper do
 
   alias ABI.TypeDecoder
   alias Explorer.Chain
-  alias Explorer.Chain.Data
+  alias Explorer.Chain.{Data, Hash}
 
   import Ecto.Query, only: [join: 5, where: 3]
   import Explorer.Chain.SmartContract, only: [burn_address_hash_string: 0]
@@ -296,4 +296,18 @@ defmodule Explorer.Helper do
   def get_app_host do
     Application.get_env(:block_scout_web, BlockScoutWeb.Endpoint)[:url][:host]
   end
+
+  @doc """
+  Converts `Explorer.Chain.Hash.t()` or string hash to DB-acceptable format.
+  For example "0xabcdef1234567890abcdef1234567890abcdef" -> "\\xabcdef1234567890abcdef1234567890abcdef"
+  """
+  @spec hash_to_query_string(Hash.t() | String.t()) :: String.t()
+  def hash_to_query_string(hash) do
+    s_hash =
+      hash
+      |> to_string()
+      |> String.trim_leading("0")
+
+    "\\#{s_hash}"
+  end
 end