Optimise graphql query (#910)

* Configure explorer processes that write to db + test function for slow query.

* Implement new query.

* Remove debug code.

* Actually use new query to fetch token txtransfers for addresses.

* Readd ordering.

* Remove 'pipeline must start with raw value' credo check because it's stupid.

.credo.exs

@@ -115,7 +115,6 @@
         {Credo.Check.Refactor.NegatedConditionsInUnless},
         {Credo.Check.Refactor.NegatedConditionsWithElse},
         {Credo.Check.Refactor.Nesting},
-        {Credo.Check.Refactor.PipeChainStart},
         {Credo.Check.Refactor.UnlessWithElse},
         {Credo.Check.Warning.BoolOperationOnSameValues},
         {Credo.Check.Warning.ExpensiveEmptyEnumCheck},

apps/block_scout_web/.iex.exs

@@ -2,5 +2,14 @@ alias Explorer.Repo
 import Ecto.Query
 
 defmodule SQLHelper do
-  def to_string(query), do: Ecto.Adapters.SQL.to_sql(:all, Explorer.Repo, query) |> IO.inspect()
+  def to_string(query), do: Ecto.Adapters.SQL.to_sql(:all, Explorer.Repo.Local, query) |> IO.inspect()
 end
+
+defmodule Clabs.Debug do
+  alias Explorer.Chain.Hash.Address
+
+  def token_tx_for_valora_address do
+    {:ok,hsh} = Address.cast("0x6131a6d616a4be3737b38988847270a64bc10caa")
+    Explorer.GraphQL.token_txtransfers_query_for_address(hsh, 26)
+  end
+end

apps/block_scout_web/lib/block_scout_web/resolvers/token_transfer_tx.ex

@@ -4,11 +4,11 @@ defmodule BlockScoutWeb.Resolvers.TokenTransferTx do
   alias Absinthe.Relay.Connection
   alias Explorer.{GraphQL, Repo}
 
-  def get_by(_, %{address_hash: address_hash} = args, _) do
+  def get_by(_, %{address_hash: address_hash, first: first} = args, _) do
     connection_args = Map.take(args, [:after, :before, :first, :last])
 
     address_hash
-    |> GraphQL.token_txtransfers_query_for_address()
+    |> GraphQL.token_txtransfers_query_for_address(first)
     |> Connection.from_query(&Repo.all/1, connection_args, options(args))
   end
 

apps/explorer/lib/explorer/application.ex

@@ -7,8 +7,6 @@ defmodule Explorer.Application do
 
   alias Explorer.{Admin, TokenTransferTokenIdMigration}
 
-  alias Explorer.Celo.Events.ContractEventStream
-
   alias Explorer.Chain.Cache.{
     Accounts,
     AddressSum,
@@ -78,38 +76,49 @@ defmodule Explorer.Application do
       configure(Explorer.ExchangeRates),
       configure(Explorer.ChainSpec.GenesisData),
       configure(Explorer.KnownTokens),
-      configure(Explorer.Market.History.Cataloger),
-      configure(Explorer.Chain.Cache.TokenExchangeRate),
-      configure(Explorer.Chain.Cache.ContractsCounter),
-      configure(Explorer.Chain.Cache.NewContractsCounter),
-      configure(Explorer.Chain.Cache.VerifiedContractsCounter),
-      configure(Explorer.Chain.Cache.NewVerifiedContractsCounter),
-      configure(Explorer.Chain.Transaction.History.Historian),
       configure(Explorer.Chain.Events.Listener),
-      configure(Explorer.Counters.AddressesWithBalanceCounter),
-      configure(Explorer.Counters.AddressesCounter),
-      configure(Explorer.Counters.AddressTransactionsCounter),
-      configure(Explorer.Counters.AddressTokenTransfersCounter),
-      configure(Explorer.Counters.AddressTransactionsGasUsageCounter),
-      configure(Explorer.Counters.AddressTokenUsdSum),
-      configure(Explorer.Counters.TokenHoldersCounter),
-      configure(Explorer.Counters.TokenTransfersCounter),
-      configure(Explorer.Counters.BlockBurnedFeeCounter),
-      configure(Explorer.Counters.BlockPriorityFeeCounter),
-      configure(Explorer.Counters.AverageBlockTime),
       configure(Explorer.Celo.AbiHandler),
       configure(Explorer.Celo.CoreContracts),
       configure(Explorer.Celo.SignerCache),
       configure(Explorer.Counters.Bridge),
       configure(Explorer.Validator.MetadataProcessor),
       configure(Explorer.Tags.AddressTag.Cataloger),
       configure(MinMissingBlockNumber),
-      configure(ContractEventStream),
       configure(TokenTransferTokenIdMigration.Supervisor)
     ]
+    |> Enum.concat(children_with_write_access())
     |> List.flatten()
   end
 
+  # child processes which write to the db after being added to the supervision tree (e.g. periodic caches, counters, etc)
+  # these cause noisy failures when connected to replica db
+  defp children_with_write_access do
+    if System.get_env("DISABLE_DB_WRITE", nil) == nil do
+      [
+        configure(Explorer.Market.History.Cataloger),
+        configure(Explorer.Chain.Cache.TokenExchangeRate),
+        configure(Explorer.Chain.Cache.ContractsCounter),
+        configure(Explorer.Chain.Cache.NewContractsCounter),
+        configure(Explorer.Chain.Cache.VerifiedContractsCounter),
+        configure(Explorer.Chain.Cache.NewVerifiedContractsCounter),
+        configure(Explorer.Chain.Transaction.History.Historian),
+        configure(Explorer.Counters.AddressesWithBalanceCounter),
+        configure(Explorer.Counters.AddressesCounter),
+        configure(Explorer.Counters.AddressTransactionsCounter),
+        configure(Explorer.Counters.AddressTokenTransfersCounter),
+        configure(Explorer.Counters.AddressTransactionsGasUsageCounter),
+        configure(Explorer.Counters.AddressTokenUsdSum),
+        configure(Explorer.Counters.TokenHoldersCounter),
+        configure(Explorer.Counters.TokenTransfersCounter),
+        configure(Explorer.Counters.BlockBurnedFeeCounter),
+        configure(Explorer.Counters.BlockPriorityFeeCounter),
+        configure(Explorer.Counters.AverageBlockTime)
+      ]
+    else
+      []
+    end
+  end
+
   defp should_start?(process) do
     Application.get_env(:explorer, process, [])[:enabled] == true
   end

apps/explorer/lib/explorer/graphql.ex

@@ -232,9 +232,66 @@ defmodule Explorer.GraphQL do
     )
   end
 
-  def token_txtransfers_query_for_address(address_hash) do
-    token_txtransfers_query()
-    |> where([t], t.to_address_hash == ^address_hash or t.from_address_hash == ^address_hash)
+  def token_txtransfers_query_for_address(address_hash, first) do
+    tt_limit = first * 2
+
+    tokens =
+      from(
+        tt in TokenTransfer,
+        where: not is_nil(tt.transaction_hash),
+        where: tt.to_address_hash == ^address_hash,
+        or_where: tt.from_address_hash == ^address_hash,
+        select: %{
+          transaction_hash: tt.transaction_hash,
+          block_number: tt.block_number,
+          to_address_hash: tt.to_address_hash,
+          from_address_hash: tt.from_address_hash
+        },
+        distinct: [desc: tt.block_number, desc: tt.transaction_hash],
+        order_by: [
+          desc: tt.block_number,
+          desc: tt.transaction_hash,
+          desc: tt.from_address_hash,
+          desc: tt.to_address_hash
+        ],
+        limit: ^tt_limit,
+        offset: 0
+      )
+
+    from(
+      tt in subquery(tokens),
+      as: :token_transfer,
+      inner_join: tx in Transaction,
+      as: :transaction,
+      on: tx.hash == tt.transaction_hash,
+      inner_join: b in Block,
+      on: tx.block_hash == b.hash,
+      left_join: token in Token,
+      on: tx.gas_currency_hash == token.contract_address_hash,
+      select: %{
+        transaction_hash: tt.transaction_hash,
+        to_address_hash: tt.to_address_hash,
+        from_address_hash: tt.from_address_hash,
+        gas_used: tx.gas_used,
+        gas_price: tx.gas_price,
+        fee_currency: tx.gas_currency_hash,
+        fee_token: fragment("coalesce(?, 'CELO')", token.symbol),
+        gateway_fee: tx.gateway_fee,
+        gateway_fee_recipient: tx.gas_fee_recipient_hash,
+        timestamp: b.timestamp,
+        input: tx.input,
+        nonce: tx.nonce,
+        block_number: tt.block_number
+      },
+      limit: ^first
+    )
+    |> order_by([transaction: t],
+      desc: t.block_number,
+      desc: t.hash,
+      asc: t.nonce,
+      desc: t.from_address_hash,
+      desc: t.to_address_hash
+    )
   end
 
   def celo_tx_transfers_query_by_txhash(tx_hash) do