feat: index miners count and total rewards from fees (#854)

* feat: index miners count and total rewards from fees

Includes a migration that takes 15-30mins to index the already
processed generation miners.

* refactor: only track miner beneficiaries

The reason for this is that miner's pks always change, so for every
new key block there will almost always be a new miner.

* test: add tests for Stats global values

* refactor: move Miners code to private functions

README.md

@@ -154,6 +154,7 @@ GET /v2/aex9/:contract_id/balances/:account_id/history - returns aex9 contract a
 
 GET /v2/deltastats                       - returns statistics for generations from tip of the chain
 GET /v2/totalstats                       - returns aggregated statistics for generations from tip of the chain
+GET /v2/minerstats                       - returns total rewards for each miner
 
 GET /v2/status                           - returns middleware status
 ```
@@ -3863,6 +3864,24 @@ $ curl -s "https://mainnet.aeternity.io/mdw/v2/totalstats?gen:421454-0&limit=1"
 
 These endpoints allows pagination, with typical `forward/backward` direction or scope denoted by `gen/from-to`.
 
+### `/v2/minerstats`
+
+Total reward given to each chain miner.
+
+```
+$ curl -s "https://mainnet.aeternity.io/mdw/v2/minerstats?limit=1" | jq '.'
+{
+  "data": [
+    {
+      "miner": "ak_2wkBCLxwjfcT3DHoisV7tGVQK8uni8XQwWZ6RUKD9DDwYSz8XN",
+      "total_reward": 76626041292504000000
+    }
+  ],
+  "next": "/v2/totalminers?cursor=ak_2wk52gAYRWAMi7gWP7A1oMvHEP9kpmp471VJFpvVzWMHnRc47a",
+  "prev": null
+}
+```
+
 ## Migrating to v2
 
 Most routes will remain the same, and can be updated by only appending the `/v2` prefix to them.

lib/ae_mdw/db/int_transfer.ex

@@ -6,6 +6,8 @@ defmodule AeMdw.Db.IntTransfer do
   alias AeMdw.Blocks
   alias AeMdw.Txs
   alias AeMdw.Db.Model
+  alias AeMdw.Db.MinerRewardsMutation
+  alias AeMdw.Db.Mutation
   alias AeMdw.Db.State
   alias AeMdw.Collection
 
@@ -21,6 +23,7 @@ defmodule AeMdw.Db.IntTransfer do
   @type target() :: Db.pubkey()
   @type ref_txi() :: Txs.txi() | -1
   @type amount() :: pos_integer()
+  @type rewards() :: [{target(), amount()}]
 
   @typep kind_suffix() :: :lock_name | :spend_name | :refund_name | :earn_oracle
 
@@ -29,26 +32,39 @@ defmodule AeMdw.Db.IntTransfer do
   @reward_block_kind "reward_block"
   @reward_dev_kind "reward_dev"
 
-  @spec block_rewards_mutation(Blocks.height(), Blocks.key_header(), Blocks.block_hash()) ::
-          IntTransfersMutation.t()
-  def block_rewards_mutation(height, key_header, key_hash) do
+  @spec block_rewards_mutations(Blocks.height(), Blocks.key_header(), Blocks.block_hash()) :: [
+          Mutation.t()
+        ]
+  def block_rewards_mutations(height, key_header, key_hash) do
     delay = :aec_governance.beneficiary_reward_delay()
     dev_benefs = Enum.map(:aec_dev_reward.beneficiaries(), &elem(&1, 0))
 
-    block_rewards =
+    {devs_rewards, miners_rewards} =
       {:node, key_header, key_hash, :key}
       |> :aec_chain_state.grant_fees(:aec_trees.new(), delay, false, nil)
       |> :aec_trees.accounts()
       |> :aeu_mtrees.to_list()
       |> Enum.map(fn {target_pk, ser_account} ->
         amount = :aec_accounts.balance(:aec_accounts.deserialize(target_pk, ser_account))
 
-        kind = (target_pk in dev_benefs && @reward_dev_kind) || @reward_block_kind
+        {target_pk, amount}
+      end)
+      |> Enum.split_with(fn {target_pk, _amount} -> target_pk in dev_benefs end)
+
+    miners_transfers =
+      Enum.map(miners_rewards, fn {target_pk, amount} ->
+        {@reward_block_kind, target_pk, amount}
+      end)
 
-        {kind, target_pk, amount}
+    devs_transfers =
+      Enum.map(devs_rewards, fn {target_pk, amount} ->
+        {@reward_dev_kind, target_pk, amount}
       end)
 
-    IntTransfersMutation.new(height, block_rewards)
+    [
+      IntTransfersMutation.new(height, miners_transfers ++ devs_transfers),
+      MinerRewardsMutation.new(miners_rewards)
+    ]
   end
 
   @spec fee(

lib/ae_mdw/db/model.ex

@@ -609,6 +609,11 @@ defmodule AeMdw.Db.Model do
 
   @type stat() :: record(:stat, index: atom(), payload: term())
 
+  @miner_defaults [:index, :total_reward]
+  defrecord :miner, @miner_defaults
+
+  @type miner() :: record(:miner, index: Db.pubkey(), total_reward: non_neg_integer())
+
   ################################################################################
 
   # starts with only chain_tables and add them progressively by groups
@@ -637,7 +642,8 @@ defmodule AeMdw.Db.Model do
       AeMdw.Db.Model.RevOrigin,
       AeMdw.Db.Model.IntTransferTx,
       AeMdw.Db.Model.KindIntTransferTx,
-      AeMdw.Db.Model.TargetKindIntTransferTx
+      AeMdw.Db.Model.TargetKindIntTransferTx,
+      AeMdw.Db.Model.Miner
     ]
   end
 
@@ -771,4 +777,5 @@ defmodule AeMdw.Db.Model do
   def record(AeMdw.Db.Model.DeltaStat), do: :delta_stat
   def record(AeMdw.Db.Model.TotalStat), do: :total_stat
   def record(AeMdw.Db.Model.Stat), do: :stat
+  def record(AeMdw.Db.Model.Miner), do: :miner
 end

lib/ae_mdw/db/mutations/miner_rewards_mutation.ex

@@ -0,0 +1,62 @@
+defmodule AeMdw.Db.MinerRewardsMutation do
+  @moduledoc """
+  Adds reward given to a miner and increases the miners count.
+  """
+
+  alias AeMdw.Db.IntTransfer
+  alias AeMdw.Db.State
+  alias AeMdw.Db.Model
+  alias AeMdw.Stats
+
+  require Model
+
+  @derive AeMdw.Db.Mutation
+  defstruct [:rewards]
+
+  @opaque t() :: %__MODULE__{rewards: IntTransfer.rewards()}
+
+  @spec new(IntTransfer.rewards()) :: t()
+  def new(rewards), do: %__MODULE__{rewards: rewards}
+
+  @spec execute(t(), State.t()) :: State.t()
+  def execute(%__MODULE__{rewards: rewards}, state) do
+    Enum.reduce(rewards, state, fn {beneficiary_pk, reward}, state ->
+      {state, new_miner?} = increment_total_reward(state, beneficiary_pk, reward)
+
+      if new_miner? do
+        increment_miners_count(state)
+      else
+        state
+      end
+    end)
+  end
+
+  defp increment_total_reward(state, beneficiary_pk, reward) do
+    case State.get(state, Model.Miner, beneficiary_pk) do
+      {:ok, Model.miner(total_reward: old_reward) = miner} ->
+        {State.put(state, Model.Miner, Model.miner(miner, total_reward: old_reward + reward)),
+         false}
+
+      :not_found ->
+        {State.put(
+           state,
+           Model.Miner,
+           Model.miner(index: beneficiary_pk, total_reward: reward)
+         ), true}
+    end
+  end
+
+  defp increment_miners_count(state) do
+    key = Stats.miners_count_key()
+
+    State.update(
+      state,
+      Model.Stat,
+      key,
+      fn
+        Model.stat(payload: count) = stat -> Model.stat(stat, payload: count + 1)
+      end,
+      Model.stat(index: key, payload: 0)
+    )
+  end
+end

lib/ae_mdw/db/sync/block.ex

@@ -91,7 +91,7 @@ defmodule AeMdw.Db.Sync.Block do
 
       block_rewards_mutation =
         if height >= AE.min_block_reward_height() do
-          IntTransfer.block_rewards_mutation(height, kb_header, kb_hash)
+          IntTransfer.block_rewards_mutations(height, kb_header, kb_hash)
         end
 
       gen_mutations = [

lib/ae_mdw/miners.ex

@@ -0,0 +1,64 @@
+defmodule AeMdw.Miners do
+  @moduledoc """
+  Context module for dealing with Miners.
+  """
+
+  alias :aeser_api_encoder, as: Enc
+  alias AeMdw.Collection
+  alias AeMdw.Db.Model
+  alias AeMdw.Db.State
+  alias AeMdw.Node.Db
+
+  require Model
+
+  @type miner() :: map()
+  @type cursor() :: binary() | nil
+  @typep state() :: State.t()
+  @typep pubkey() :: Db.pubkey()
+  @typep pagination() :: Collection.direction_limit()
+
+  @spec fetch_miners(state(), pagination(), cursor()) :: {cursor(), [miner()], cursor()}
+  def fetch_miners(state, pagination, cursor) do
+    cursor = deserialize_cursor(cursor)
+
+    {prev_cursor, miners, next_cursor} =
+      state
+      |> build_streamer(cursor)
+      |> Collection.paginate(pagination)
+
+    {serialize_cursor(prev_cursor), Enum.map(miners, &fetch_miner!(state, &1)),
+     serialize_cursor(next_cursor)}
+  end
+
+  @spec fetch_miner!(state(), pubkey()) :: miner()
+  def fetch_miner!(state, miner_pk),
+    do: render_miner(State.fetch!(state, Model.Miner, miner_pk))
+
+  defp build_streamer(state, cursor), do: &Collection.stream(state, Model.Miner, &1, nil, cursor)
+
+  defp render_miner(
+         Model.miner(
+           index: miner_pk,
+           total_reward: total_reward
+         )
+       ) do
+    %{
+      miner: Enc.encode(:account_pubkey, miner_pk),
+      total_reward: total_reward
+    }
+  end
+
+  defp serialize_cursor(nil), do: nil
+
+  defp serialize_cursor({miner_pk, is_reversed?}),
+    do: {Enc.encode(:account_pubkey, miner_pk), is_reversed?}
+
+  defp deserialize_cursor(nil), do: nil
+
+  defp deserialize_cursor(cursor_bin) do
+    case Enc.safe_decode(:account_pubkey, cursor_bin) do
+      {:ok, miner_pk} -> {:ok, miner_pk}
+      {:error, _reason} -> nil
+    end
+  end
+end

lib/ae_mdw/stats.ex

@@ -29,6 +29,7 @@ defmodule AeMdw.Stats do
   @typep range() :: {:gen, Range.t()} | nil
 
   @tps_stat_key :max_tps
+  @miners_count_stat_key :miners_count
 
   @spec mutation(height(), Db.key_block(), [Db.micro_block()], txi(), txi(), boolean()) ::
           StatsMutation.t()
@@ -54,6 +55,9 @@ defmodule AeMdw.Stats do
   @spec max_tps_key() :: atom()
   def max_tps_key, do: @tps_stat_key
 
+  @spec miners_count_key() :: atom()
+  def miners_count_key, do: @miners_count_stat_key
+
   # Legacy v1 is a blending between /totalstats and /deltastats.
   # The active and inactive object counters are totals while the rewards are delta.
   @spec fetch_stats_v1(State.t(), direction(), range(), cursor(), limit()) ::
@@ -126,14 +130,17 @@ defmodule AeMdw.Stats do
 
   @spec fetch_stats(State.t()) :: {:ok, map()} | {:error, Error.t()}
   def fetch_stats(state) do
-    case State.get(state, Model.Stat, @tps_stat_key) do
-      {:ok, Model.stat(payload: {tps, tps_block_hash})} ->
-        {:ok,
-         %{
-           max_transactions_per_second: tps,
-           max_transactions_per_second_block_hash: Enc.encode(:key_block_hash, tps_block_hash)
-         }}
-
+    with {:ok, Model.stat(payload: {tps, tps_block_hash})} <-
+           State.get(state, Model.Stat, @tps_stat_key),
+         {:ok, Model.stat(payload: miners_count)} <-
+           State.get(state, Model.Stat, @miners_count_stat_key) do
+      {:ok,
+       %{
+         max_transactions_per_second: tps,
+         max_transactions_per_second_block_hash: Enc.encode(:key_block_hash, tps_block_hash),
+         miners_count: miners_count
+       }}
+    else
       :not_found ->
         {:error, ErrInput.NotFound.exception(value: "no stats")}
     end

lib/ae_mdw_web/controllers/stats_controller.ex

@@ -1,6 +1,7 @@
 defmodule AeMdwWeb.StatsController do
   use AeMdwWeb, :controller
 
+  alias AeMdw.Miners
   alias AeMdw.Stats
   alias AeMdwWeb.Plugs.PaginatedPlug
   alias AeMdwWeb.FallbackController
@@ -62,4 +63,13 @@ defmodule AeMdwWeb.StatsController do
       {:error, reason} -> {:error, reason}
     end
   end
+
+  @spec miners(Conn.t(), map()) :: Conn.t()
+  def miners(%Conn{assigns: assigns} = conn, _params) do
+    %{state: state, pagination: pagination, cursor: cursor} = assigns
+
+    {prev_cursor, miners, next_cursor} = Miners.fetch_miners(state, pagination, cursor)
+
+    Util.paginate(conn, prev_cursor, miners, next_cursor)
+  end
 end

lib/ae_mdw_web/router.ex

@@ -79,6 +79,7 @@ defmodule AeMdwWeb.Router do
 
       get "/deltastats", StatsController, :delta_stats
       get "/stats", StatsController, :stats
+      get "/minerstats", StatsController, :miners
 
       scope "/swagger" do
         forward "/", SwaggerForwardV2,