chore: make the wealth rank task work with the database instead of AsyncStore and ets

lib/ae_mdw/application.ex

@@ -191,7 +191,6 @@ defmodule AeMdw.Application do
 
     AeMdw.Sync.AsyncTasks.Stats.init()
     AeMdw.Sync.AsyncTasks.Store.init()
-    AeMdw.Sync.AsyncTasks.WealthRank.init()
 
     AeMdw.Db.AsyncStore.init()
     AeMdw.Sync.Aex9BalancesCache.init()

lib/ae_mdw/db/model.ex

@@ -92,6 +92,16 @@ defmodule AeMdw.Db.Model do
             index: {amount(), pubkey()}
           )
 
+  @account_balance_defaults [index: nil, balance: 0]
+  defrecord :account_balance, @account_balance_defaults
+
+  @type account_balance_index() :: pubkey()
+  @type account_balance() ::
+          record(:account_balance,
+            index: account_balance_index(),
+            balance: non_neg_integer()
+          )
+
   # txs block index :
   #     index = {kb_index (0..), mb_index}, tx_index = tx_index, hash = block (header) hash
   #     On keyblock boundary: mb_index = -1}
@@ -1322,13 +1332,15 @@ defmodule AeMdw.Db.Model do
   defp tasks_tables() do
     [
       AeMdw.Db.Model.BalanceAccount,
+      AeMdw.Db.Model.AccountBalance,
       AeMdw.Db.Model.AsyncTask,
       AeMdw.Db.Model.Migrations
     ]
   end
 
   @spec record(atom()) :: atom()
   def record(AeMdw.Db.Model.BalanceAccount), do: :balance_account
+  def record(AeMdw.Db.Model.AccountBalance), do: :account_balance
   def record(AeMdw.Db.Model.AsyncTask), do: :async_task
   def record(AeMdw.Db.Model.Migrations), do: :migrations
   def record(AeMdw.Db.Model.Tx), do: :tx

lib/ae_mdw/db/mutations/update_balance_accounts_mutation.ex

@@ -0,0 +1,68 @@
+defmodule AeMdw.Db.UpdateBalanceAccountMutation do
+  @moduledoc """
+    Update the balance of an account.
+  """
+
+  alias AeMdw.Db.Model
+  alias AeMdw.Db.State
+  alias AeMdw.Node.Db
+
+  require Model
+
+  @derive AeMdw.Db.Mutation
+  defstruct [:account_pk, :balance]
+
+  @typep account_balance() :: non_neg_integer()
+  @opaque t() :: %__MODULE__{
+            account_pk: Db.pubkey(),
+            balance: account_balance()
+          }
+
+  @spec new(Db.pubkey(), account_balance()) :: t()
+  def new(account_pk, balance) do
+    %__MODULE__{
+      account_pk: account_pk,
+      balance: balance
+    }
+  end
+
+  @spec execute(t(), State.t()) :: State.t()
+  def execute(
+        %__MODULE__{
+          account_pk: account_pk,
+          balance: balance
+        },
+        state
+      ) do
+    state
+    |> get_balance(account_pk)
+    |> case do
+      old_balance when old_balance != nil ->
+        state
+        |> State.delete(Model.BalanceAccount, {old_balance, account_pk})
+        |> State.delete(Model.AccountBalance, account_pk)
+
+      _balance ->
+        state
+    end
+    |> insert(account_pk, balance)
+  end
+
+  defp insert(state, pubkey, balance) do
+    balance_account_record = Model.balance_account(index: {balance, pubkey})
+    account_balance_record = Model.account_balance(index: pubkey, balance: balance)
+
+    state
+    |> State.put(Model.BalanceAccount, balance_account_record)
+    |> State.put(Model.AccountBalance, account_balance_record)
+  end
+
+  defp get_balance(state, pubkey) do
+    state
+    |> State.get(Model.AccountBalance, pubkey)
+    |> case do
+      {:ok, Model.account_balance(balance: balance)} -> balance
+      :not_found -> nil
+    end
+  end
+end

lib/ae_mdw/sync/async_store_server.ex

@@ -13,7 +13,6 @@ defmodule AeMdw.Sync.AsyncStoreServer do
   alias AeMdw.Db.AsyncStore
   alias AeMdw.Db.Mutation
   alias AeMdw.Db.State
-  alias AeMdw.Sync.AsyncTasks.WealthRank
 
   @typep internal_state() :: %{async_store: AsyncStore.t(), last_db_kbi: AeMdw.Blocks.height()}
 
@@ -65,17 +64,13 @@ defmodule AeMdw.Sync.AsyncStoreServer do
         _from,
         %{async_store: async_store} = state
       ) do
-    {top_keys, store} = WealthRank.prune_balance_ranking(async_store)
-
     db_state2 =
-      store
+      async_store
       |> AsyncStore.mutations()
       |> Enum.reduce(db_state1, &Mutation.execute/2)
 
     AsyncStore.clear(async_store)
 
-    _store = WealthRank.restore_ranking(async_store, top_keys)
-
     {:reply, db_state2, state}
   end
 end

lib/ae_mdw/sync/async_tasks/consumer.ex

@@ -120,7 +120,13 @@ defmodule AeMdw.Sync.AsyncTasks.Consumer do
   defp process(Model.async_task(index: {_ts, type} = index, args: args, extra_args: extra_args)) do
     mod = @type_mod[type]
     done_fn = fn -> Producer.notify_consumed(index, args) end
-    mod.process(args ++ extra_args, done_fn)
+
+    if mod == StoreAccountBalance do
+      Logger.info("Skipping #{inspect(index)}")
+    else
+      mod.process(args ++ extra_args, done_fn)
+    end
+
     :ok
   end