Luke Imhoff
6 years ago
committed by
GitHub
commit
9cf9583165
@ -0,0 +1,672 @@ |
||||
defmodule Explorer.Chain.Import do |
||||
@moduledoc """ |
||||
Bulk importing of data into `Explorer.Repo` |
||||
""" |
||||
|
||||
import Ecto.Query, only: [from: 2] |
||||
|
||||
alias Ecto.{Changeset, Multi} |
||||
alias Explorer.Chain.{Address, Balance, Block, Hash, InternalTransaction, Log, Transaction, Wei} |
||||
alias Explorer.Repo |
||||
|
||||
@typep addresses_option :: {:addresses, [params_option | timeout_option | with_option]} |
||||
@typep balances_option :: {:balances, [params_option | timeout_option]} |
||||
@typep blocks_option :: {:blocks, [params_option | timeout_option]} |
||||
@typep broadcast_option :: {:broadcast, Boolean} |
||||
@typep internal_transactions_option :: {:internal_transactions, [params_option | timeout_option]} |
||||
@typep logs_option :: {:logs, [params_option | timeout_option]} |
||||
@typep on_conflict_option :: {:on_conflict, :nothing | :replace_all} |
||||
@typep params_option :: {:params, [map()]} |
||||
@typep receipts_option :: {:receipts, [params_option | timeout_option]} |
||||
@typep timeout_option :: {:timeout, timeout} |
||||
@typep timestamps :: %{inserted_at: DateTime.t(), updated_at: DateTime.t()} |
||||
@typep timestamps_option :: {:timestamps, timestamps} |
||||
@typep transactions_option :: {:transactions, [on_conflict_option | params_option | timeout_option | with_option]} |
||||
@typep with_option :: {:with, changeset_function_name :: atom} |
||||
|
||||
@type all_options :: [ |
||||
addresses_option |
||||
| balances_option |
||||
| blocks_option |
||||
| broadcast_option |
||||
| internal_transactions_option |
||||
| logs_option |
||||
| receipts_option |
||||
| timeout_option |
||||
| transactions_option |
||||
] |
||||
@type all_result :: |
||||
{:ok, |
||||
%{ |
||||
optional(:addresses) => [Address.t()], |
||||
optional(:balances) => [ |
||||
%{required(:address_hash) => Hash.Address.t(), required(:block_number) => Block.block_number()} |
||||
], |
||||
optional(:blocks) => [Block.t()], |
||||
optional(:internal_transactions) => [ |
||||
%{required(:index) => non_neg_integer(), required(:transaction_hash) => Hash.Full.t()} |
||||
], |
||||
optional(:logs) => [Log.t()], |
||||
optional(:receipts) => [Hash.Full.t()], |
||||
optional(:transactions) => [Hash.Full.t()] |
||||
}} |
||||
| {:error, [Changeset.t()]} |
||||
| {:error, step :: Ecto.Multi.name(), failed_value :: any(), |
||||
changes_so_far :: %{optional(Ecto.Multi.name()) => any()}} |
||||
|
||||
@type internal_transactions_options :: [ |
||||
addresses_option |
||||
| internal_transactions_option |
||||
| timeout_option |
||||
| {:transactions, [{:hashes, [String.t()]} | timeout_option]} |
||||
] |
||||
@type internal_transactions_result :: |
||||
{:ok, |
||||
%{ |
||||
optional(:addresses) => [Hash.Address.t()], |
||||
optional(:internal_transactions) => [ |
||||
%{required(:index) => non_neg_integer(), required(:transaction_hash) => Hash.Full.t()} |
||||
] |
||||
}} |
||||
| {:error, [Changeset.t()]} |
||||
| {:error, step :: Ecto.Multi.name(), failed_value :: any(), |
||||
changes_so_far :: %{optional(Ecto.Multi.name()) => any()}} |
||||
|
||||
# timeouts all in milliseconds |
||||
|
||||
@transaction_timeout 120_000 |
||||
|
||||
@insert_addresses_timeout 60_000 |
||||
@insert_balances_timeout 60_000 |
||||
@insert_blocks_timeout 60_000 |
||||
@insert_internal_transactions_timeout 60_000 |
||||
@insert_logs_timeout 60_000 |
||||
@insert_transactions_timeout 60_000 |
||||
|
||||
@doc """ |
||||
Bulk insert all data stored in the `Explorer`. |
||||
|
||||
The import returns the unique key(s) for each type of record inserted. |
||||
|
||||
| Key | Value Type | Value Description | |
||||
|--------------------------|-------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------| |
||||
| `:addresses` | `[Explorer.Chain.Address.t()]` | List of `t:Explorer.Chain.Address.t/0`s | |
||||
| `:balances` | `[%{address_hash: Explorer.Chain.Hash.t(), block_number: Explorer.Chain.Block.block_number()}]` | List of `t:Explorer.Chain.Address.t/0`s | |
||||
| `:blocks` | `[Explorer.Chain.Block.t()]` | List of `t:Explorer.Chain.Block.t/0`s | |
||||
| `:internal_transactions` | `[%{index: non_neg_integer(), transaction_hash: Explorer.Chain.Hash.t()}]` | List of maps of the `t:Explorer.Chain.InternalTransaction.t/0` `index` and `transaction_hash` | |
||||
| `:logs` | `[Explorer.Chain.Log.t()]` | List of `t:Explorer.Chain.Log.t/0`s | |
||||
| `:transactions` | `[Explorer.Chain.Hash.t()]` | List of `t:Explorer.Chain.Transaction.t/0` `hash` | |
||||
|
||||
The params for each key are validated using the corresponding `Ecto.Schema` module's `changeset/2` function. If there |
||||
are errors, they are returned in `Ecto.Changeset.t`s, so that the original, invalid value can be reconstructed for any |
||||
error messages. |
||||
|
||||
Because there are multiple processes potentially writing to the same tables at the same time, |
||||
`c:Ecto.Repo.insert_all/2`'s |
||||
[`:conflict_target` and `:on_conflict` options](https://hexdocs.pm/ecto/Ecto.Repo.html#c:insert_all/3-options) are |
||||
used to perform [upserts](https://hexdocs.pm/ecto/Ecto.Repo.html#c:insert_all/3-upserts) on all tables, so that |
||||
a pre-existing unique key will not trigger a failure, but instead replace or otherwise update the row. |
||||
|
||||
## Data Notifications |
||||
|
||||
On successful inserts, processes interested in certain domains of data will be notified |
||||
that new data has been inserted. See `Explorer.Chain.subscribe_to_events/1` for more information. |
||||
|
||||
## Options |
||||
|
||||
* `:addresses` |
||||
* `:params` - `list` of params for `Explorer.Chain.Address.changeset/2`. |
||||
* `:timeout` - the timeout for inserting all addresses. Defaults to `#{@insert_addresses_timeout}` milliseconds. |
||||
* `:with` - the changeset function on `Explorer.Chain.Address` to use validate `:params`. |
||||
* `:balances` |
||||
* `:params` - `list` of params for `Explorer.Chain.Balance.changeset/2`. |
||||
* `:timeout` - the timeout for inserting all balances. Defaults to `#{@insert_balances_timeout}` milliseconds. |
||||
* `:blocks` |
||||
* `:params` - `list` of params for `Explorer.Chain.Block.changeset/2`. |
||||
* `:timeout` - the timeout for inserting all blocks. Defaults to `#{@insert_blocks_timeout}` milliseconds. |
||||
* `:broacast` - Boolean flag indicating whether or not to broadcast the event. |
||||
* `:internal_transactions` |
||||
* `:params` - `list` of params for `Explorer.Chain.InternalTransaction.changeset/2`. |
||||
* `:timeout` - the timeout for inserting all internal transactions. Defaults to |
||||
`#{@insert_internal_transactions_timeout}` milliseconds. |
||||
* `:logs` |
||||
* `:params` - `list` of params for `Explorer.Chain.Log.changeset/2`. |
||||
* `:timeout` - the timeout for inserting all logs. Defaults to `#{@insert_logs_timeout}` milliseconds. |
||||
* `:timeout` - the timeout for the whole `c:Ecto.Repo.transaction/0` call. Defaults to `#{@transaction_timeout}` |
||||
milliseconds. |
||||
* `:transactions` |
||||
* `:on_conflict` - Whether to do `:nothing` or `:replace_all` columns when there is a pre-existing transaction |
||||
with the same hash. |
||||
|
||||
*NOTE*: Because the repository transaction for a pending `Explorer.Chain.Transaction`s could `COMMIT` after the |
||||
repository transaction for that same transaction being collated into a block, writers, it is recomended to use |
||||
`:nothing` for pending transactions and `:replace_all` for collated transactions, so that collated transactions |
||||
win. |
||||
* `:params` - `list` of params for `Explorer.Chain.Transaction.changeset/2`. |
||||
* `:timeout` - the timeout for inserting all transactions found in the params lists across all |
||||
types. Defaults to `#{@insert_transactions_timeout}` milliseconds. |
||||
* `:with` - the changeset function on `Explorer.Chain.Transaction` to use validate `:params`. |
||||
""" |
||||
@spec all(all_options()) :: all_result() |
||||
def all(options) when is_list(options) do |
||||
broadcast = |
||||
case Keyword.fetch(options, :broadcast) do |
||||
{:ok, broadcast} -> broadcast |
||||
:error -> false |
||||
end |
||||
|
||||
changes_list_arguments_list = import_options_to_changes_list_arguments_list(options) |
||||
|
||||
with {:ok, ecto_schema_module_to_changes_list_map} <- |
||||
changes_list_arguments_list_to_ecto_schema_module_to_changes_list_map(changes_list_arguments_list), |
||||
{:ok, data} <- insert_ecto_schema_module_to_changes_list_map(ecto_schema_module_to_changes_list_map, options) do |
||||
if broadcast, do: broadcast_events(data) |
||||
{:ok, data} |
||||
end |
||||
end |
||||
|
||||
defp broadcast_events(data) do |
||||
for {event_type, event_data} <- data, event_type in ~w(addresses balances blocks logs transactions)a do |
||||
broadcast_event_data(event_type, event_data) |
||||
end |
||||
end |
||||
|
||||
defp broadcast_event_data(event_type, event_data) do |
||||
Registry.dispatch(Registry.ChainEvents, event_type, fn entries -> |
||||
for {pid, _registered_val} <- entries do |
||||
send(pid, {:chain_event, event_type, event_data}) |
||||
end |
||||
end) |
||||
end |
||||
|
||||
defp changes_list_arguments_list_to_ecto_schema_module_to_changes_list_map(changes_list_arguments_list) do |
||||
changes_list_arguments_list |
||||
|> Stream.map(fn [params_list, options] -> |
||||
ecto_schema_module = Keyword.fetch!(options, :for) |
||||
{ecto_schema_module, changes_list(params_list, options)} |
||||
end) |
||||
|> Enum.reduce({:ok, %{}}, fn |
||||
{ecto_schema_module, {:ok, changes_list}}, {:ok, ecto_schema_module_to_changes_list_map} -> |
||||
{:ok, Map.put(ecto_schema_module_to_changes_list_map, ecto_schema_module, changes_list)} |
||||
|
||||
{_, {:ok, _}}, {:error, _} = error -> |
||||
error |
||||
|
||||
{_, {:error, _} = error}, {:ok, _} -> |
||||
error |
||||
|
||||
{_, {:error, changesets}}, {:error, acc_changesets} -> |
||||
{:error, acc_changesets ++ changesets} |
||||
end) |
||||
end |
||||
|
||||
@spec changes_list(params :: [map], [{:for, module} | {:with, atom}]) :: {:ok, [map]} | {:error, [Changeset.t()]} |
||||
defp changes_list(params, options) when is_list(options) do |
||||
ecto_schema_module = Keyword.fetch!(options, :for) |
||||
changeset_function_name = Keyword.get(options, :with, :changeset) |
||||
struct = ecto_schema_module.__struct__() |
||||
|
||||
{status, acc} = |
||||
params |
||||
|> Stream.map(&apply(ecto_schema_module, changeset_function_name, [struct, &1])) |
||||
|> Enum.reduce({:ok, []}, fn |
||||
changeset = %Changeset{valid?: false}, {:ok, _} -> |
||||
{:error, [changeset]} |
||||
|
||||
changeset = %Changeset{valid?: false}, {:error, acc_changesets} -> |
||||
{:error, [changeset | acc_changesets]} |
||||
|
||||
%Changeset{changes: changes, valid?: true}, {:ok, acc_changes} -> |
||||
{:ok, [changes | acc_changes]} |
||||
|
||||
%Changeset{valid?: true}, {:error, _} = error -> |
||||
error |
||||
end) |
||||
|
||||
{status, Enum.reverse(acc)} |
||||
end |
||||
|
||||
@import_option_key_to_ecto_schema_module %{ |
||||
addresses: Address, |
||||
balances: Balance, |
||||
blocks: Block, |
||||
internal_transactions: InternalTransaction, |
||||
logs: Log, |
||||
transactions: Transaction |
||||
} |
||||
|
||||
defp ecto_schema_module_to_changes_list_map_to_multi(ecto_schema_module_to_changes_list_map, options) |
||||
when is_list(options) do |
||||
timestamps = timestamps() |
||||
full_options = Keyword.put(options, :timestamps, timestamps) |
||||
|
||||
Multi.new() |
||||
|> run_addresses(ecto_schema_module_to_changes_list_map, full_options) |
||||
|> run_balances(ecto_schema_module_to_changes_list_map, full_options) |
||||
|> run_blocks(ecto_schema_module_to_changes_list_map, full_options) |
||||
|> run_transactions(ecto_schema_module_to_changes_list_map, full_options) |
||||
|> run_internal_transactions(ecto_schema_module_to_changes_list_map, full_options) |
||||
|> run_logs(ecto_schema_module_to_changes_list_map, full_options) |
||||
end |
||||
|
||||
defp run_addresses(multi, ecto_schema_module_to_changes_list_map, options) |
||||
when is_map(ecto_schema_module_to_changes_list_map) and is_list(options) do |
||||
case ecto_schema_module_to_changes_list_map do |
||||
%{Address => addresses_changes} -> |
||||
timestamps = Keyword.fetch!(options, :timestamps) |
||||
|
||||
Multi.run(multi, :addresses, fn _ -> |
||||
insert_addresses( |
||||
addresses_changes, |
||||
timeout: options[:addresses][:timeout] || @insert_addresses_timeout, |
||||
timestamps: timestamps |
||||
) |
||||
end) |
||||
|
||||
_ -> |
||||
multi |
||||
end |
||||
end |
||||
|
||||
defp run_balances(multi, ecto_schema_module_to_changes_list_map, options) |
||||
when is_map(ecto_schema_module_to_changes_list_map) and is_list(options) do |
||||
case ecto_schema_module_to_changes_list_map do |
||||
%{Balance => balances_changes} -> |
||||
timestamps = Keyword.fetch!(options, :timestamps) |
||||
|
||||
Multi.run(multi, :balances, fn _ -> |
||||
insert_balances( |
||||
balances_changes, |
||||
timeout: options[:balances][:timeout] || @insert_balances_timeout, |
||||
timestamps: timestamps |
||||
) |
||||
end) |
||||
|
||||
_ -> |
||||
multi |
||||
end |
||||
end |
||||
|
||||
defp run_blocks(multi, ecto_schema_module_to_changes_list_map, options) |
||||
when is_map(ecto_schema_module_to_changes_list_map) and is_list(options) do |
||||
case ecto_schema_module_to_changes_list_map do |
||||
%{Block => blocks_changes} -> |
||||
timestamps = Keyword.fetch!(options, :timestamps) |
||||
|
||||
Multi.run(multi, :blocks, fn _ -> |
||||
insert_blocks( |
||||
blocks_changes, |
||||
timeout: options[:blocks][:timeout] || @insert_blocks_timeout, |
||||
timestamps: timestamps |
||||
) |
||||
end) |
||||
|
||||
_ -> |
||||
multi |
||||
end |
||||
end |
||||
|
||||
defp run_transactions(multi, ecto_schema_module_to_changes_list_map, options) |
||||
when is_map(ecto_schema_module_to_changes_list_map) and is_list(options) do |
||||
case ecto_schema_module_to_changes_list_map do |
||||
%{Transaction => transactions_changes} -> |
||||
# check required options as early as possible |
||||
transactions_options = Keyword.fetch!(options, :transactions) |
||||
on_conflict = Keyword.fetch!(transactions_options, :on_conflict) |
||||
timestamps = Keyword.fetch!(options, :timestamps) |
||||
|
||||
Multi.run(multi, :transactions, fn _ -> |
||||
insert_transactions( |
||||
transactions_changes, |
||||
on_conflict: on_conflict, |
||||
timeout: transactions_options[:timeout] || @insert_transactions_timeout, |
||||
timestamps: timestamps |
||||
) |
||||
end) |
||||
|
||||
_ -> |
||||
multi |
||||
end |
||||
end |
||||
|
||||
defp run_internal_transactions(multi, ecto_schema_module_to_changes_list_map, options) |
||||
when is_map(ecto_schema_module_to_changes_list_map) and is_list(options) do |
||||
case ecto_schema_module_to_changes_list_map do |
||||
%{InternalTransaction => internal_transactions_changes} -> |
||||
timestamps = Keyword.fetch!(options, :timestamps) |
||||
|
||||
multi |
||||
|> Multi.run(:internal_transactions, fn _ -> |
||||
insert_internal_transactions( |
||||
internal_transactions_changes, |
||||
timeout: options[:internal_transactions][:timeout] || @insert_internal_transactions_timeout, |
||||
timestamps: timestamps |
||||
) |
||||
end) |
||||
|> Multi.run(:internal_transactions_indexed_at_transactions, fn %{internal_transactions: internal_transactions} |
||||
when is_list(internal_transactions) -> |
||||
update_transactions_internal_transactions_indexed_at( |
||||
internal_transactions, |
||||
timeout: options[:transactions][:timeout] || @insert_transactions_timeout, |
||||
timestamps: timestamps |
||||
) |
||||
end) |
||||
|
||||
_ -> |
||||
multi |
||||
end |
||||
end |
||||
|
||||
defp run_logs(multi, ecto_schema_module_to_changes_list_map, options) |
||||
when is_map(ecto_schema_module_to_changes_list_map) and is_list(options) do |
||||
case ecto_schema_module_to_changes_list_map do |
||||
%{Log => logs_changes} -> |
||||
timestamps = Keyword.fetch!(options, :timestamps) |
||||
|
||||
Multi.run(multi, :logs, fn _ -> |
||||
insert_logs( |
||||
logs_changes, |
||||
timeout: options[:logs][:timeout] || @insert_logs_timeout, |
||||
timestamps: timestamps |
||||
) |
||||
end) |
||||
|
||||
_ -> |
||||
multi |
||||
end |
||||
end |
||||
|
||||
@spec insert_addresses([%{hash: Hash.Address.t()}], [timeout_option | timestamps_option | with_option]) :: |
||||
{:ok, [Hash.Address.t()]} |
||||
defp insert_addresses(changes_list, named_arguments) |
||||
when is_list(changes_list) and is_list(named_arguments) do |
||||
timestamps = Keyword.fetch!(named_arguments, :timestamps) |
||||
timeout = Keyword.fetch!(named_arguments, :timeout) |
||||
|
||||
# order so that row ShareLocks are grabbed in a consistent order |
||||
ordered_changes_list = sort_address_changes_list(changes_list) |
||||
|
||||
insert_changes_list( |
||||
ordered_changes_list, |
||||
conflict_target: :hash, |
||||
on_conflict: |
||||
from( |
||||
address in Address, |
||||
update: [ |
||||
set: [ |
||||
contract_code: fragment("COALESCE(?, EXCLUDED.contract_code)", address.contract_code), |
||||
# ARGMAX on two columns |
||||
fetched_balance: |
||||
fragment( |
||||
""" |
||||
CASE WHEN EXCLUDED.fetched_balance_block_number IS NOT NULL AND |
||||
(? IS NULL OR |
||||
EXCLUDED.fetched_balance_block_number >= ?) THEN |
||||
EXCLUDED.fetched_balance |
||||
ELSE ? |
||||
END |
||||
""", |
||||
address.fetched_balance_block_number, |
||||
address.fetched_balance_block_number, |
||||
address.fetched_balance |
||||
), |
||||
# MAX on two columns |
||||
fetched_balance_block_number: |
||||
fragment( |
||||
""" |
||||
CASE WHEN EXCLUDED.fetched_balance_block_number IS NOT NULL AND |
||||
(? IS NULL OR |
||||
EXCLUDED.fetched_balance_block_number >= ?) THEN |
||||
EXCLUDED.fetched_balance_block_number |
||||
ELSE ? |
||||
END |
||||
""", |
||||
address.fetched_balance_block_number, |
||||
address.fetched_balance_block_number, |
||||
address.fetched_balance_block_number |
||||
) |
||||
] |
||||
] |
||||
), |
||||
for: Address, |
||||
returning: true, |
||||
timeout: timeout, |
||||
timestamps: timestamps |
||||
) |
||||
end |
||||
|
||||
defp sort_address_changes_list(changes_list) do |
||||
Enum.sort_by(changes_list, & &1.hash) |
||||
end |
||||
|
||||
@spec insert_balances( |
||||
[ |
||||
%{ |
||||
required(:address_hash) => Hash.Address.t(), |
||||
required(:block_number) => Block.block_number(), |
||||
required(:value) => Wei.t() |
||||
} |
||||
], |
||||
[timeout_option] |
||||
) :: |
||||
{:ok, [%{required(:address_hash) => Hash.Address.t(), required(:block_number) => Block.block_number()}]} |
||||
| {:error, [Changeset.t()]} |
||||
defp insert_balances(changes_list, named_arguments) when is_list(changes_list) and is_list(named_arguments) do |
||||
timestamps = Keyword.fetch!(named_arguments, :timestamps) |
||||
timeout = Keyword.fetch!(named_arguments, :timeout) |
||||
|
||||
# order so that row ShareLocks are grabbed in a consistent order |
||||
ordered_changes_list = Enum.sort_by(changes_list, &{&1.address_hash, &1.block_number}) |
||||
|
||||
{:ok, _} = |
||||
insert_changes_list( |
||||
ordered_changes_list, |
||||
conflict_target: [:address_hash, :block_number], |
||||
on_conflict: |
||||
from( |
||||
balance in Balance, |
||||
update: [ |
||||
set: [ |
||||
inserted_at: fragment("LEAST(EXCLUDED.inserted_at, ?)", balance.inserted_at), |
||||
updated_at: fragment("GREATEST(EXCLUDED.updated_at, ?)", balance.updated_at), |
||||
value: |
||||
fragment( |
||||
""" |
||||
CASE WHEN EXCLUDED.updated_at > ? THEN EXCLUDED.value |
||||
ELSE ? |
||||
END |
||||
""", |
||||
balance.updated_at, |
||||
balance.value |
||||
) |
||||
] |
||||
] |
||||
), |
||||
for: Balance, |
||||
timeout: timeout, |
||||
timestamps: timestamps |
||||
) |
||||
|
||||
{:ok, Enum.map(ordered_changes_list, &Map.take(&1, ~w(address_hash block_number)a))} |
||||
end |
||||
|
||||
@spec insert_blocks([map()], [timeout_option | timestamps_option]) :: {:ok, [Block.t()]} | {:error, [Changeset.t()]} |
||||
defp insert_blocks(changes_list, named_arguments) |
||||
when is_list(changes_list) and is_list(named_arguments) do |
||||
timestamps = Keyword.fetch!(named_arguments, :timestamps) |
||||
timeout = Keyword.fetch!(named_arguments, :timeout) |
||||
|
||||
# order so that row ShareLocks are grabbed in a consistent order |
||||
ordered_changes_list = Enum.sort_by(changes_list, &{&1.number, &1.hash}) |
||||
|
||||
{:ok, blocks} = |
||||
insert_changes_list( |
||||
ordered_changes_list, |
||||
conflict_target: :number, |
||||
on_conflict: :replace_all, |
||||
for: Block, |
||||
returning: true, |
||||
timeout: timeout, |
||||
timestamps: timestamps |
||||
) |
||||
|
||||
{:ok, blocks} |
||||
end |
||||
|
||||
@spec insert_internal_transactions([map], [timeout_option | timestamps_option]) :: |
||||
{:ok, [%{index: non_neg_integer, transaction_hash: Hash.t()}]} |
||||
| {:error, [Changeset.t()]} |
||||
defp insert_internal_transactions(changes_list, named_arguments) |
||||
when is_list(changes_list) and is_list(named_arguments) do |
||||
timestamps = Keyword.fetch!(named_arguments, :timestamps) |
||||
|
||||
# order so that row ShareLocks are grabbed in a consistent order |
||||
ordered_changes_list = Enum.sort_by(changes_list, &{&1.transaction_hash, &1.index}) |
||||
|
||||
{:ok, internal_transactions} = |
||||
insert_changes_list( |
||||
ordered_changes_list, |
||||
conflict_target: [:transaction_hash, :index], |
||||
for: InternalTransaction, |
||||
on_conflict: :replace_all, |
||||
returning: [:index, :transaction_hash], |
||||
timestamps: timestamps |
||||
) |
||||
|
||||
{:ok, |
||||
for( |
||||
internal_transaction <- internal_transactions, |
||||
do: Map.take(internal_transaction, [:index, :transaction_hash]) |
||||
)} |
||||
end |
||||
|
||||
@spec insert_logs([map()], [timeout_option | timestamps_option]) :: |
||||
{:ok, [Log.t()]} |
||||
| {:error, [Changeset.t()]} |
||||
defp insert_logs(changes_list, named_arguments) |
||||
when is_list(changes_list) and is_list(named_arguments) do |
||||
timestamps = Keyword.fetch!(named_arguments, :timestamps) |
||||
timeout = Keyword.fetch!(named_arguments, :timeout) |
||||
|
||||
# order so that row ShareLocks are grabbed in a consistent order |
||||
ordered_changes_list = Enum.sort_by(changes_list, &{&1.transaction_hash, &1.index}) |
||||
|
||||
{:ok, _} = |
||||
insert_changes_list( |
||||
ordered_changes_list, |
||||
conflict_target: [:transaction_hash, :index], |
||||
on_conflict: :replace_all, |
||||
for: Log, |
||||
returning: true, |
||||
timeout: timeout, |
||||
timestamps: timestamps |
||||
) |
||||
end |
||||
|
||||
@spec insert_transactions([map()], [on_conflict_option | timeout_option | timestamps_option]) :: |
||||
{:ok, [Hash.t()]} | {:error, [Changeset.t()]} |
||||
defp insert_transactions(changes_list, named_arguments) |
||||
when is_list(changes_list) and is_list(named_arguments) do |
||||
timestamps = Keyword.fetch!(named_arguments, :timestamps) |
||||
timeout = Keyword.fetch!(named_arguments, :timeout) |
||||
on_conflict = Keyword.fetch!(named_arguments, :on_conflict) |
||||
|
||||
# order so that row ShareLocks are grabbed in a consistent order |
||||
ordered_changes_list = Enum.sort_by(changes_list, & &1.hash) |
||||
|
||||
{:ok, transactions} = |
||||
insert_changes_list( |
||||
ordered_changes_list, |
||||
conflict_target: :hash, |
||||
on_conflict: on_conflict, |
||||
for: Transaction, |
||||
returning: [:hash], |
||||
timeout: timeout, |
||||
timestamps: timestamps |
||||
) |
||||
|
||||
{:ok, for(transaction <- transactions, do: transaction.hash)} |
||||
end |
||||
|
||||
defp insert_changes_list(changes_list, options) when is_list(changes_list) do |
||||
ecto_schema_module = Keyword.fetch!(options, :for) |
||||
|
||||
timestamped_changes_list = timestamp_changes_list(changes_list, Keyword.fetch!(options, :timestamps)) |
||||
|
||||
{_, inserted} = |
||||
Repo.safe_insert_all( |
||||
ecto_schema_module, |
||||
timestamped_changes_list, |
||||
Keyword.delete(options, :for) |
||||
) |
||||
|
||||
{:ok, inserted} |
||||
end |
||||
|
||||
defp update_transactions_internal_transactions_indexed_at(internal_transactions, named_arguments) |
||||
when is_list(internal_transactions) and is_list(named_arguments) do |
||||
timeout = Keyword.fetch!(named_arguments, :timeout) |
||||
timestamps = Keyword.fetch!(named_arguments, :timestamps) |
||||
|
||||
ordered_transaction_hashes = |
||||
internal_transactions |
||||
|> MapSet.new(& &1.transaction_hash) |
||||
|> Enum.sort() |
||||
|
||||
query = |
||||
from( |
||||
t in Transaction, |
||||
where: t.hash in ^ordered_transaction_hashes, |
||||
update: [set: [internal_transactions_indexed_at: ^timestamps.updated_at]] |
||||
) |
||||
|
||||
transaction_count = Enum.count(ordered_transaction_hashes) |
||||
|
||||
{^transaction_count, result} = Repo.update_all(query, [], timeout: timeout) |
||||
|
||||
{:ok, result} |
||||
end |
||||
|
||||
defp timestamp_changes_list(changes_list, timestamps) when is_list(changes_list) do |
||||
Enum.map(changes_list, ×tamp_params(&1, timestamps)) |
||||
end |
||||
|
||||
defp timestamp_params(changes, timestamps) when is_map(changes) do |
||||
Map.merge(changes, timestamps) |
||||
end |
||||
|
||||
defp import_options_to_changes_list_arguments_list(options) do |
||||
Enum.flat_map(@import_option_key_to_ecto_schema_module, fn {option_key, ecto_schema_module} -> |
||||
case Keyword.fetch(options, option_key) do |
||||
{:ok, option_value} when is_list(option_value) -> |
||||
[ |
||||
[ |
||||
Keyword.fetch!(option_value, :params), |
||||
[for: ecto_schema_module, with: Keyword.get(option_value, :with, :changeset)] |
||||
] |
||||
] |
||||
|
||||
:error -> |
||||
[] |
||||
end |
||||
end) |
||||
end |
||||
|
||||
defp import_transaction(multi, options) when is_list(options) do |
||||
Repo.transaction(multi, timeout: Keyword.get(options, :timeout, @transaction_timeout)) |
||||
end |
||||
|
||||
defp insert_ecto_schema_module_to_changes_list_map(ecto_schema_module_to_changes_list_map, options) do |
||||
timestamps = timestamps() |
||||
|
||||
ecto_schema_module_to_changes_list_map |
||||
|> ecto_schema_module_to_changes_list_map_to_multi(Keyword.put(options, :timestamps, timestamps)) |
||||
|> import_transaction(options) |
||||
end |
||||
|
||||
@spec timestamps() :: timestamps |
||||
defp timestamps do |
||||
now = DateTime.utc_now() |
||||
%{inserted_at: now, updated_at: now} |
||||
end |
||||
end |
||||
@ -0,0 +1,117 @@ |
||||
defmodule Explorer.Chain.ImportTest do |
||||
use Explorer.DataCase |
||||
|
||||
alias Explorer.Chain.Import |
||||
|
||||
doctest Import |
||||
|
||||
describe "all/1" do |
||||
test "updates address with contract code" do |
||||
smart_contract_bytecode = |
||||
"0x608060405234801561001057600080fd5b5060df8061001f6000396000f3006080604052600436106049576000357c0100000000000000000000000000000000000000000000000000000000900463ffffffff16806360fe47b114604e5780636d4ce63c146078575b600080fd5b348015605957600080fd5b5060766004803603810190808035906020019092919050505060a0565b005b348015608357600080fd5b50608a60aa565b6040518082815260200191505060405180910390f35b8060008190555050565b600080549050905600a165627a7a7230582040d82a7379b1ee1632ad4d8a239954fd940277b25628ead95259a85c5eddb2120029" |
||||
|
||||
address_hash = "0x1c494fa496f1cfd918b5ff190835af3aaf60987e" |
||||
insert(:address, hash: address_hash) |
||||
|
||||
from_address_hash = "0x8cc2e4b51b4340cb3727cffe3f1878756e732cee" |
||||
from_address = insert(:address, hash: from_address_hash) |
||||
|
||||
transaction_string_hash = "0x0705ea0a5b997d9aafd5c531e016d9aabe3297a28c0bd4ef005fe6ea329d301b" |
||||
insert(:transaction, from_address: from_address, hash: transaction_string_hash) |
||||
|
||||
options = [ |
||||
addresses: [ |
||||
params: [ |
||||
%{ |
||||
contract_code: smart_contract_bytecode, |
||||
hash: address_hash |
||||
} |
||||
] |
||||
], |
||||
internal_transactions: [ |
||||
params: [ |
||||
%{ |
||||
created_contract_address_hash: address_hash, |
||||
created_contract_code: smart_contract_bytecode, |
||||
from_address_hash: from_address_hash, |
||||
gas: 184_531, |
||||
gas_used: 84531, |
||||
index: 0, |
||||
init: |
||||
"0x6060604052341561000c57fe5b5b6101a68061001c6000396000f300606060405263ffffffff7c01000000000000000000000000000000000000000000000000000000006000350416631d3b9edf811461005b57806366098d4f1461007b578063a12f69e01461009b578063f4f3bdc1146100bb575bfe5b6100696004356024356100db565b60408051918252519081900360200190f35b61006960043560243561010a565b60408051918252519081900360200190f35b610069600435602435610124565b60408051918252519081900360200190f35b610069600435602435610163565b60408051918252519081900360200190f35b60008282028315806100f757508284828115156100f457fe5b04145b15156100ff57fe5b8091505b5092915050565b6000828201838110156100ff57fe5b8091505b5092915050565b60008080831161013057fe5b828481151561013b57fe5b049050828481151561014957fe5b0681840201841415156100ff57fe5b8091505b5092915050565b60008282111561016f57fe5b508082035b929150505600a165627a7a7230582020c944d8375ca14e2c92b14df53c2d044cb99dc30c3ba9f55e2bcde87bd4709b0029", |
||||
trace_address: [], |
||||
transaction_hash: transaction_string_hash, |
||||
type: "create", |
||||
value: 0 |
||||
} |
||||
] |
||||
] |
||||
] |
||||
|
||||
assert {:ok, _} = Import.all(options) |
||||
|
||||
address = Explorer.Repo.one(from(address in Explorer.Chain.Address, where: address.hash == ^address_hash)) |
||||
|
||||
assert address.contract_code != nil |
||||
end |
||||
|
||||
test "with internal_transactions updates Transaction internal_transactions_indexed_at" do |
||||
from_address_hash = "0xe8ddc5c7a2d2f0d7a9798459c0104fdf5e987aca" |
||||
to_address_hash = "0x8bf38d4764929064f2d4d3a56520a76ab3df415b" |
||||
transaction_hash = "0x3a3eb134e6792ce9403ea4188e5e79693de9e4c94e499db132be086400da79e6" |
||||
|
||||
options = [ |
||||
addresses: [ |
||||
params: [ |
||||
%{hash: from_address_hash}, |
||||
%{hash: to_address_hash} |
||||
] |
||||
], |
||||
transactions: [ |
||||
params: [ |
||||
%{ |
||||
from_address_hash: from_address_hash, |
||||
gas: 4_677_320, |
||||
gas_price: 1, |
||||
hash: transaction_hash, |
||||
input: "0x", |
||||
nonce: 0, |
||||
r: 0, |
||||
s: 0, |
||||
v: 0, |
||||
value: 0 |
||||
} |
||||
], |
||||
on_conflict: :replace_all |
||||
], |
||||
internal_transactions: [ |
||||
params: [ |
||||
%{ |
||||
block_number: 35, |
||||
call_type: "call", |
||||
from_address_hash: from_address_hash, |
||||
gas: 4_677_320, |
||||
gas_used: 27770, |
||||
index: 0, |
||||
output: "0x", |
||||
to_address_hash: to_address_hash, |
||||
trace_address: [], |
||||
transaction_hash: transaction_hash, |
||||
type: "call", |
||||
value: 0 |
||||
} |
||||
] |
||||
] |
||||
] |
||||
|
||||
refute Enum.any?(options[:transactions][:params], &Map.has_key?(&1, :internal_transactions_indexed_at)) |
||||
|
||||
assert {:ok, _} = Import.all(options) |
||||
|
||||
transaction = |
||||
Explorer.Repo.one(from(transaction in Explorer.Chain.Transaction, where: transaction.hash == ^transaction_hash)) |
||||
|
||||
refute transaction.internal_transactions_indexed_at == nil |
||||
end |
||||
end |
||||
end |
||||
Loading…
Reference in new issue