add block_hash to token_transfers

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

@@ -8,7 +8,7 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
   import Ecto.Query, only: [from: 2, subquery: 1]
 
   alias Ecto.{Changeset, Multi, Repo}
-  alias Explorer.Chain.{Address, Block, Import, InternalTransaction, Log, TokenTransfer, Transaction}
+  alias Explorer.Chain.{Address, Block, Import, InternalTransaction, Log, Transaction}
   alias Explorer.Chain.Block.Reward
   alias Explorer.Chain.Import.Runner
   alias Explorer.Chain.Import.Runner.Address.CurrentTokenBalances
@@ -92,9 +92,6 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
     |> Multi.run(:acquire_contract_address_tokens, fn repo, _ ->
       acquire_contract_address_tokens(repo, consensus_block_numbers)
     end)
-    |> Multi.run(:remove_nonconsensus_token_transfers, fn repo, %{derive_transaction_forks: transactions} ->
-      remove_nonconsensus_token_transfers(repo, transactions, insert_options)
-    end)
     |> Multi.run(:delete_address_token_balances, fn repo, _ ->
       delete_address_token_balances(repo, consensus_block_numbers, insert_options)
     end)
@@ -317,35 +314,6 @@ defmodule Explorer.Chain.Import.Runner.Blocks do
       {:error, %{exception: postgrex_error, consensus_block_numbers: consensus_block_numbers}}
   end
 
-  defp remove_nonconsensus_token_transfers(repo, forked_transaction_hashes, %{timeout: timeout}) do
-    ordered_token_transfers =
-      from(
-        token_transfer in TokenTransfer,
-        where: token_transfer.transaction_hash in ^forked_transaction_hashes,
-        select: token_transfer.transaction_hash,
-        # Enforce TokenTransfer ShareLocks order (see docs: sharelocks.md)
-        order_by: [
-          token_transfer.transaction_hash,
-          token_transfer.log_index
-        ],
-        lock: "FOR UPDATE"
-      )
-
-    query =
-      from(token_transfer in TokenTransfer,
-        select: map(token_transfer, [:transaction_hash, :log_index]),
-        inner_join: ordered_token_transfer in subquery(ordered_token_transfers),
-        on: ordered_token_transfer.transaction_hash == token_transfer.transaction_hash
-      )
-
-    {_count, deleted_token_transfers} = repo.delete_all(query, timeout: timeout)
-
-    {:ok, deleted_token_transfers}
-  rescue
-    postgrex_error in Postgrex.Error ->
-      {:error, %{exception: postgrex_error, transactions: forked_transaction_hashes}}
-  end
-
   defp remove_nonconsensus_internal_transactions(repo, forked_transaction_hashes, %{timeout: timeout}) do
     query =
       from(

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

@@ -61,7 +61,7 @@ defmodule Explorer.Chain.Import.Runner.TokenTransfers do
       Import.insert_changes_list(
         repo,
         ordered_changes_list,
-        conflict_target: [:transaction_hash, :log_index],
+        conflict_target: [:transaction_hash, :log_index, :block_hash],
         on_conflict: on_conflict,
         for: TokenTransfer,
         returning: true,

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

@@ -27,7 +27,7 @@ defmodule Explorer.Chain.TokenTransfer do
   import Ecto.Changeset
   import Ecto.Query, only: [from: 2, limit: 2, where: 3]
 
-  alias Explorer.Chain.{Address, Hash, TokenTransfer, Transaction}
+  alias Explorer.Chain.{Address, Block, Hash, TokenTransfer, Transaction}
   alias Explorer.Chain.Token.Instance
   alias Explorer.{PagingOptions, Repo}
 
@@ -35,6 +35,7 @@ defmodule Explorer.Chain.TokenTransfer do
 
   @typedoc """
   * `:amount` - The token transferred amount
+  * `:block_hash` - hash of the block
   * `:block_number` - The block number that the transfer took place.
   * `:from_address` - The `t:Explorer.Chain.Address.t/0` that sent the tokens
   * `:from_address_hash` - Address hash foreign key
@@ -50,6 +51,7 @@ defmodule Explorer.Chain.TokenTransfer do
   @type t :: %TokenTransfer{
           amount: Decimal.t(),
           block_number: non_neg_integer() | nil,
+          block_hash: Hash.Full.t(),
           from_address: %Ecto.Association.NotLoaded{} | Address.t(),
           from_address_hash: Hash.Address.t(),
           to_address: %Ecto.Association.NotLoaded{} | Address.t(),
@@ -93,6 +95,13 @@ defmodule Explorer.Chain.TokenTransfer do
       type: Hash.Full
     )
 
+    belongs_to(:block, Block,
+      foreign_key: :block_hash,
+      primary_key: true,
+      references: :hash,
+      type: Hash.Full
+    )
+
     has_one(
       :instance,
       Instance,

apps/explorer/priv/repo/migrations/20191122062035_add_block_hash_to_token_transfers.exs

@@ -0,0 +1,33 @@
+defmodule Explorer.Repo.Migrations.AddBlockHashToTokenTransfers do
+  use Ecto.Migration
+
+  def change do
+    alter table(:token_transfers) do
+      add(:block_hash, :bytea)
+    end
+
+    execute("""
+    UPDATE token_transfers token_transfer
+    SET block_hash = with_block.block_hash
+    FROM (
+      SELECT transfer.transaction_hash,
+      t.block_hash
+      FROM token_transfers transfer
+      JOIN transactions t
+      ON t.hash = transfer.transaction_hash
+    ) AS with_block
+    WHERE token_transfer.transaction_hash = with_block.transaction_hash
+    ;
+    """)
+
+    alter table(:token_transfers) do
+      modify(:block_hash, references(:blocks, column: :hash, type: :bytea), null: false)
+    end
+
+    execute("""
+    ALTER table token_transfers
+    DROP CONSTRAINT token_transfers_pkey,
+    ADD PRIMARY KEY (transaction_hash, block_hash, log_index);
+    """)
+  end
+end

apps/indexer/lib/indexer/transform/token_transfers.ex

@@ -40,6 +40,7 @@ defmodule Indexer.Transform.TokenTransfers do
     token_transfer = %{
       amount: Decimal.new(amount || 0),
       block_number: log.block_number,
+      block_hash: log.block_hash,
       log_index: log.index,
       from_address_hash: truncate_address_hash(log.second_topic),
       to_address_hash: truncate_address_hash(log.third_topic),
@@ -64,6 +65,7 @@ defmodule Indexer.Transform.TokenTransfers do
     token_transfer = %{
       block_number: log.block_number,
       log_index: log.index,
+      block_hash: log.block_hash,
       from_address_hash: truncate_address_hash(log.second_topic),
       to_address_hash: truncate_address_hash(log.third_topic),
       token_contract_address_hash: log.address_hash,
@@ -87,6 +89,7 @@ defmodule Indexer.Transform.TokenTransfers do
 
     token_transfer = %{
       block_number: log.block_number,
+      block_hash: log.block_hash,
       log_index: log.index,
       from_address_hash: encode_address_hash(from_address_hash),
       to_address_hash: encode_address_hash(to_address_hash),