Update acquire_contract_address_tokens to mitigate deadlocks

(cherry picked from commit 9f089cc4adba9b678e0c88fa2b0e3ad3c5830f17)

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

@@ -21,19 +21,70 @@ defmodule Explorer.Chain.Import.Runner.Tokens do
   @type holder_count :: non_neg_integer()
   @type token_holder_count :: %{contract_address_hash: Hash.Address.t(), count: holder_count()}
 
-  def acquire_contract_address_tokens(repo, contract_address_hashes) do
-    token_query =
-      from(
-        token in Token,
-        where: token.contract_address_hash in ^contract_address_hashes,
-        # Enforce Token ShareLocks order (see docs: sharelocks.md)
-        order_by: [
-          token.contract_address_hash
-        ],
-        lock: "FOR UPDATE"
+  def acquire_contract_address_tokens(repo, contract_address_hashes_and_token_ids) do
+    initial_query_no_token_id =
+      from(token in Token,
+        select: token
+      )
+
+    initial_query_with_token_id =
+      from(token in Token,
+        left_join: instance in Token.Instance,
+        on: token.contract_address_hash == instance.token_contract_address_hash,
+        select: token
       )
 
-    tokens = repo.all(token_query)
+    {query_no_token_id, query_with_token_id} =
+      contract_address_hashes_and_token_ids
+      |> Enum.reduce({initial_query_no_token_id, initial_query_with_token_id}, fn {contract_address_hash, token_id},
+                                                                                  {query_no_token_id,
+                                                                                   query_with_token_id} ->
+        if is_nil(token_id) do
+          {from(
+             token in query_no_token_id,
+             or_where: token.contract_address_hash == ^contract_address_hash
+           ), query_with_token_id}
+        else
+          {query_no_token_id,
+           from(
+             [token, instance] in query_with_token_id,
+             or_where: token.contract_address_hash == ^contract_address_hash and instance.token_id == ^token_id
+           )}
+        end
+      end)
+
+    final_query_no_token_id =
+      if query_no_token_id == initial_query_no_token_id do
+        nil
+      else
+        from(
+          token in query_no_token_id,
+          # Enforce Token ShareLocks order (see docs: sharelocks.md)
+          order_by: [
+            token.contract_address_hash
+          ],
+          lock: "FOR UPDATE"
+        )
+      end
+
+    final_query_with_token_id =
+      if query_with_token_id == initial_query_with_token_id do
+        nil
+      else
+        from(
+          [token, instance] in query_with_token_id,
+          # Enforce Token ShareLocks order (see docs: sharelocks.md)
+          order_by: [
+            token.contract_address_hash,
+            instance.token_id
+          ],
+          lock: "FOR UPDATE"
+        )
+      end
+
+    tokens_no_token_id = (final_query_no_token_id && repo.all(final_query_no_token_id)) || []
+    tokens_with_token_id = (final_query_with_token_id && repo.all(final_query_with_token_id)) || []
+    tokens = tokens_no_token_id ++ tokens_with_token_id
 
     {:ok, tokens}
   end