feat: add top miners endpoint

lib/ae_mdw/db/int_transfer.ex

@@ -8,6 +8,7 @@ defmodule AeMdw.Db.IntTransfer do
   alias AeMdw.Db.Model
   alias AeMdw.Db.MinerRewardsMutation
   alias AeMdw.Db.Mutation
+  alias AeMdw.Db.TopMinerStatsMutation
   alias AeMdw.Db.State
   alias AeMdw.Collection
 
@@ -40,6 +41,7 @@ defmodule AeMdw.Db.IntTransfer do
   def block_rewards_mutations(key_block) do
     height = :aec_blocks.height(key_block)
     delay = :aec_governance.beneficiary_reward_delay()
+    time = :aec_blocks.time_in_msecs(key_block)
 
     dev_benefs =
       for {protocol, _height} <- :aec_hard_forks.protocols(),
@@ -72,7 +74,8 @@ defmodule AeMdw.Db.IntTransfer do
 
     [
       IntTransfersMutation.new(height, miners_transfers ++ devs_transfers),
-      MinerRewardsMutation.new(miners_rewards)
+      MinerRewardsMutation.new(miners_rewards),
+      TopMinerStatsMutation.new(miners_rewards, time)
     ]
   end
 

lib/ae_mdw/db/model.ex

@@ -1236,6 +1236,14 @@ defmodule AeMdw.Db.Model do
   @type miner_index() :: pubkey()
   @type miner() :: record(:miner, index: miner_index(), total_reward: non_neg_integer())
 
+  @type top_miner_stats_index() ::
+          {Stats.interval_by(), Stats.interval_start(), pos_integer(), pubkey()}
+  @type top_miner_stats() ::
+          record(:top_miner_stats, index: top_miner_stats_index())
+
+  @top_miner_stats_defaults [:index]
+  defrecord :top_miner_stats, @top_miner_stats_defaults
+
   ################################################################################
 
   # starts with only chain_tables and add them progressively by groups
@@ -1373,7 +1381,8 @@ defmodule AeMdw.Db.Model do
       AeMdw.Db.Model.DeltaStat,
       AeMdw.Db.Model.TotalStat,
       AeMdw.Db.Model.Stat,
-      AeMdw.Db.Model.Statistic
+      AeMdw.Db.Model.Statistic,
+      AeMdw.Db.Model.TopMinerStats
     ]
   end
 
@@ -1486,4 +1495,5 @@ defmodule AeMdw.Db.Model do
   def record(AeMdw.Db.Model.Statistic), do: :statistic
   def record(AeMdw.Db.Model.Miner), do: :miner
   def record(AeMdw.Db.Model.AccountNamesCount), do: :account_names_count
+  def record(AeMdw.Db.Model.TopMinerStats), do: :top_miner_stats
 end

lib/ae_mdw/db/mutations/top_miner_stats_mutation.ex

@@ -0,0 +1,70 @@
+defmodule AeMdw.Db.TopMinerStatsMutation do
+  @moduledoc """
+  Increments the top miners stats.
+  """
+
+  alias AeMdw.Collection
+  alias AeMdw.Db.IntTransfer
+  alias AeMdw.Db.State
+  alias AeMdw.Db.Model
+  alias AeMdw.Db.Sync.Stats
+
+  require Model
+
+  @derive AeMdw.Db.Mutation
+  defstruct [:rewards, :time]
+
+  @opaque t() :: %__MODULE__{rewards: IntTransfer.rewards(), time: non_neg_integer()}
+
+  @spec new(IntTransfer.rewards(), non_neg_integer()) :: t()
+  def new(rewards, time), do: %__MODULE__{rewards: rewards, time: time}
+
+  @spec execute(t(), State.t()) :: State.t()
+  def execute(%__MODULE__{rewards: rewards, time: time}, state) do
+    Enum.reduce(rewards, state, fn {beneficiary_pk, _reward}, state ->
+      increment_top_miners(state, time, beneficiary_pk)
+    end)
+  end
+
+  defp increment_top_miners(state, time, beneficiary_pk) do
+    time
+    |> Stats.time_intervals()
+    |> Enum.reduce(state, fn {interval_by, interval_start}, state ->
+      kb =
+        Collection.generate_key_boundary(
+          {interval_by, interval_start, Collection.integer(), Collection.binary()}
+        )
+
+      state
+      |> Collection.stream(Model.TopMinerStats, :backward, kb, nil)
+      |> Stream.filter(fn {_interval_by, _interval_start, _count, bpk} ->
+        bpk == beneficiary_pk
+      end)
+      |> tap(&IO.inspect(Enum.count(&1)))
+      |> Enum.at(0, :none)
+      |> case do
+        {^interval_by, ^interval_start, count, ^beneficiary_pk} ->
+          IO.inspect("updating")
+
+          state
+          |> State.delete(
+            Model.TopMinerStats,
+            {interval_by, interval_start, count, beneficiary_pk}
+          )
+          |> State.put(
+            Model.TopMinerStats,
+            Model.top_miner_stats(index: {interval_by, interval_start, count + 1, beneficiary_pk})
+          )
+
+        :none ->
+          IO.inspect("inserting missing")
+
+          State.put(
+            state,
+            Model.TopMinerStats,
+            Model.top_miner_stats(index: {interval_by, interval_start, 1, beneficiary_pk})
+          )
+      end
+    end)
+  end
+end

lib/ae_mdw/stats.ex

@@ -280,6 +280,24 @@ defmodule AeMdw.Stats do
     end
   end
 
+  @spec fetch_top_miners_stats(State.t(), pagination(), query(), range(), cursor()) ::
+          {:ok, {pagination_cursor(), [statistic()], pagination_cursor()}} | {:error, reason()}
+  def fetch_top_miners_stats(state, pagination, query, range, cursor) do
+    with {:ok, filters} <- Util.convert_params(query, &convert_param/1),
+         {:ok, cursor} <- deserialize_top_miners_cursor(cursor) do
+      paginated_top_miners =
+        state
+        |> build_top_miners_streamer(filters, range, cursor)
+        |> Collection.paginate(
+          pagination,
+          &render_top_miner_statistic(state, &1),
+          &serialize_top_miners_cursor/1
+        )
+
+      {:ok, paginated_top_miners}
+    end
+  end
+
   defp fetch_statistics(state, pagination, filters, range, cursor, tag) do
     with {:ok, cursor} <- deserialize_statistic_cursor(cursor) do
       paginated_statistics =
@@ -315,6 +333,21 @@ defmodule AeMdw.Stats do
     end
   end
 
+  defp build_top_miners_streamer(state, filters, _scope, cursor) do
+    interval_by = Map.get(filters, :interval_by, :day)
+    {start_network_date, end_network_date} = DbUtil.network_date_interval(state)
+    min_date = filters |> Map.get(:min_start_date, start_network_date) |> to_interval(interval_by)
+    max_date = filters |> Map.get(:max_start_date, end_network_date) |> to_interval(interval_by)
+
+    key_boundary =
+      {{interval_by, min_date, 0, Util.min_bin()},
+       {interval_by, max_date, Util.max_int(), Util.max_256bit_bin()}}
+
+    fn direction ->
+      Collection.stream(state, Model.TopMinerStats, direction, key_boundary, cursor)
+    end
+  end
+
   defp fill_missing_dates(stream, tag, interval_by, :backward, cursor, min_date, max_date) do
     max_date =
       case cursor do
@@ -401,6 +434,42 @@ defmodule AeMdw.Stats do
     render_statistic(state, {:virtual, statistic_key, count})
   end
 
+  defp render_top_miner_statistic(
+         _state,
+         {:month, interval_start, count, beneficiary_id}
+       ) do
+    %{
+      start_date: months_to_iso(interval_start),
+      end_date: months_to_iso(interval_start + 1),
+      miner: :aeapi.format_account_pubkey(beneficiary_id),
+      count: count
+    }
+  end
+
+  defp render_top_miner_statistic(
+         _state,
+         {:week, interval_start, count, beneficiary_id}
+       ) do
+    %{
+      start_date: days_to_iso(interval_start * @days_per_week),
+      end_date: days_to_iso((interval_start + 1) * @days_per_week),
+      miner: :aeapi.format_account_pubkey(beneficiary_id),
+      count: count
+    }
+  end
+
+  defp render_top_miner_statistic(
+         _state,
+         {:day, interval_start, count, beneficiary_id}
+       ) do
+    %{
+      start_date: days_to_iso(interval_start),
+      end_date: days_to_iso(interval_start + 1),
+      miner: :aeapi.format_account_pubkey(beneficiary_id),
+      count: count
+    }
+  end
+
   defp convert_blocks_param({"type", "key"}), do: {:ok, {:block_type, :key}}
   defp convert_blocks_param({"type", "micro"}), do: {:ok, {:block_type, :micro}}
   defp convert_blocks_param(param), do: convert_param(param)
@@ -440,6 +509,12 @@ defmodule AeMdw.Stats do
   defp serialize_statistics_cursor({:virtual, {_tag, _interval_by, interval_start}, _count}),
     do: "#{interval_start}"
 
+  defp serialize_top_miners_cursor({_interval_by, _interval_start, _count, _ben} = cursor) do
+    cursor
+    |> :erlang.term_to_binary()
+    |> Base.encode64()
+  end
+
   defp deserialize_statistic_cursor(nil), do: {:ok, nil}
 
   defp deserialize_statistic_cursor(cursor_bin) do
@@ -449,6 +524,21 @@ defmodule AeMdw.Stats do
     end
   end
 
+  defp deserialize_top_miners_cursor(nil), do: {:ok, nil}
+
+  defp deserialize_top_miners_cursor(cursor_bin) do
+    case Base.decode64(cursor_bin) do
+      {:ok, bin} ->
+        case :erlang.binary_to_term(bin) do
+          cursor when is_tuple(cursor) -> {:ok, cursor}
+          _ -> {:error, ErrInput.Cursor.exception(value: cursor_bin)}
+        end
+
+      :error ->
+        {:error, ErrInput.Cursor.exception(value: cursor_bin)}
+    end
+  end
+
   defp render_delta_stats(state, gens), do: Enum.map(gens, &fetch_delta_stat!(state, &1))
 
   defp render_total_stats(state, gens), do: Enum.map(gens, &fetch_total_stat!(state, &1))

lib/ae_mdw_web/controllers/stats_controller.ex

@@ -134,4 +134,14 @@ defmodule AeMdwWeb.StatsController do
       Util.render(conn, paginated_stats)
     end
   end
+
+  @spec top_miners_stats(Conn.t(), map()) :: Conn.t()
+  def top_miners_stats(%Conn{assigns: assigns} = conn, _params) do
+    %{state: state, pagination: pagination, query: query, scope: scope, cursor: cursor} = assigns
+
+    with {:ok, paginated_stats} <-
+           Stats.fetch_top_miners_stats(state, pagination, query, scope, cursor) do
+      Util.render(conn, paginated_stats)
+    end
+  end
 end

lib/ae_mdw_web/router.ex

@@ -73,6 +73,7 @@ defmodule AeMdwWeb.Router do
       get "/stats/total", StatsController, :total_stats
       get "/stats/delta", StatsController, :delta_stats
       get "/stats/miners", StatsController, :miners_stats
+      get "/stats/miners/top", StatsController, :top_miners_stats
       get "/stats/contracts", StatsController, :contracts_stats
       get "/stats/aex9-transfers", StatsController, :aex9_transfers_stats
       get "/stats", StatsController, :stats

priv/migrations/20241025101739_generate_top_miners.ex

@@ -0,0 +1,43 @@
+defmodule AeMdw.Migrations.GenerateTopMiners do
+  alias AeMdw.Db.TopMinerStatsMutation
+  alias AeMdw.Db.Model
+  alias AeMdw.Db.RocksDbCF
+  alias AeMdw.Db.State
+
+  require Model
+
+  @spec run(State.t(), boolean()) :: {:ok, non_neg_integer()}
+  def run(state, _from_start?) do
+    # dev_benefs =
+    #   for {protocol, _height} <- :aec_hard_forks.protocols(),
+    #       {pk, _share} <- :aec_dev_reward.beneficiaries(protocol) do
+    #     pk
+    #   end
+
+    # delay = :aec_governance.beneficiary_reward_delay()
+
+    {_state, count} =
+      Model.Block
+      |> RocksDbCF.stream()
+      |> Stream.filter(fn Model.block(index: {_key_index, micro_index}) -> micro_index == -1 end)
+      |> Stream.map(fn Model.block() = block ->
+        Model.block(hash: hash) = block
+        {:ok, key_block} = :aec_chain.get_block(hash)
+        time = :aec_blocks.time_in_msecs(key_block)
+
+        miner =
+          key_block
+          |> :aec_blocks.to_header()
+          |> :aec_headers.miner()
+
+        TopMinerStatsMutation.new([{miner, 0}], time)
+      end)
+      |> Stream.chunk_every(1000)
+      |> Enum.reduce({state, 0}, fn mutations, {state, count} ->
+        len = length(mutations)
+        {State.commit_db(state, mutations), count + len}
+      end)
+
+    {:ok, count}
+  end
+end

test/ae_mdw/db/sync/block_test.exs

@@ -1,7 +1,9 @@
 defmodule AeMdw.Db.Sync.BlockTest do
   use ExUnit.Case
 
+  alias AeMdw.Collection
   alias AeMdw.Db.Model
+  alias AeMdw.Db.State
   alias AeMdw.Db.Sync.Block
 
   import AeMdwWeb.BlockchainSim,