Schedule sending of reward tokens to the node reward address (#477)

* Fix unspent output mapping

* Send node rewards after mint_rewards

* Refactor

* Add test

* Fix subset test error

lib/archethic/mining/pending_transaction_validation.ex

@@ -112,7 +112,7 @@ defmodule Archethic.Mining.PendingTransactionValidation do
            }
          }
        }) do
-    case Reward.get_transfers(Reward.last_scheduling_date()) do
+    case Reward.get_transfers() do
       ^token_transfers ->
         :ok
 

lib/archethic/mining/validation_context.ex

@@ -721,7 +721,7 @@ defmodule Archethic.Mining.ValidationContext do
     initial_movements =
       tx
       |> Transaction.get_movements()
-      |> Enum.map(&{&1.to, &1})
+      |> Enum.map(&{{&1.to, &1.type}, &1})
       |> Enum.into(%{})
 
     fee =
@@ -731,14 +731,18 @@ defmodule Archethic.Mining.ValidationContext do
       )
 
     resolved_movements =
-      Enum.reduce(resolved_addresses, [], fn {to, resolved}, acc ->
-        case Map.get(initial_movements, to) do
-          nil ->
-            acc
+      Enum.reduce(resolved_addresses, [], fn
+        {to, resolved}, acc ->
+          case Map.get(initial_movements, to) do
+            nil ->
+              acc
 
-          movement ->
-            [%{movement | to: resolved} | acc]
-        end
+            movement ->
+              [%{movement | to: resolved} | acc]
+          end
+
+        _, acc ->
+          acc
       end)
 
     resolved_recipients =

lib/archethic/reward.ex

@@ -9,28 +9,36 @@ defmodule Archethic.Reward do
 
   alias Archethic.Election
 
+  alias Archethic.Account
+
+  alias Archethic.SharedSecrets
+
   alias Archethic.P2P
   alias Archethic.P2P.Node
 
   alias __MODULE__.Scheduler
 
   alias Archethic.TransactionChain.Transaction
   alias Archethic.TransactionChain.TransactionData
-  alias Archethic.TransactionChain.TransactionData.UCOLedger.Transfer
+  alias Archethic.TransactionChain.TransactionData.Ledger
+  alias Archethic.TransactionChain.TransactionData.TokenLedger
+  alias Archethic.TransactionChain.TransactionData.TokenLedger.Transfer
 
   @unit_uco 100_000_000
 
   @doc """
-  Get the minimum rewards for validation nodes
+  Get rewards amount for validation nodes
   """
-  @spec min_validation_nodes_reward() :: pos_integer()
-  def min_validation_nodes_reward do
-    uco_eur_price =
-      DateTime.utc_now()
+  @spec validation_nodes_reward() :: pos_integer()
+  def validation_nodes_reward do
+    date = DateTime.utc_now()
+
+    uco_usd_price =
+      date
       |> OracleChain.get_uco_price()
-      |> Keyword.get(:eur)
+      |> Keyword.get(:usd)
 
-    trunc(uco_eur_price * 50) * @unit_uco
+    trunc(50 / uco_usd_price / Calendar.ISO.days_in_month(date.year, date.month) * @unit_uco)
   end
 
   @doc """
@@ -61,34 +69,105 @@ defmodule Archethic.Reward do
     Transaction.new(:mint_rewards, data)
   end
 
+  @spec new_node_rewards() :: Transaction.t()
+  def new_node_rewards() do
+    data = %TransactionData{
+      ledger: %Ledger{
+        token: %TokenLedger{
+          transfers: get_transfers()
+        }
+      }
+    }
+
+    Transaction.new(:node_rewards, data)
+  end
+
   @doc """
   Determine if the local node is the initiator of the new rewards mint
   """
-  @spec initiator?() :: boolean()
-  def initiator?(index \\ 0) do
+  @spec initiator?(binary()) :: boolean()
+  def initiator?(address, index \\ 0) do
     %Node{first_public_key: initiator_key} =
-      next_address()
+      address
       |> Election.storage_nodes(P2P.authorized_and_available_nodes())
       |> Enum.at(index)
 
     initiator_key == Crypto.first_node_public_key()
   end
 
-  defp next_address do
+  @spec next_address() :: binary()
+  def next_address do
     key_index = Crypto.number_of_network_pool_keys()
     next_public_key = Crypto.network_pool_public_key(key_index + 1)
     Crypto.derive_address(next_public_key)
   end
 
   @doc """
-  Return the list of transfers to rewards the validation nodes for a specific date
+  Return the list of transfers to rewards the validation nodes
   """
-  @spec get_transfers(last_reward_date :: DateTime.t()) :: reward_transfers :: list(Transfer.t())
-  def get_transfers(_last_date = %DateTime{}) do
-    # TODO
-    []
+  @spec get_transfers() :: reward_transfers :: list(Transfer.t())
+  def get_transfers() do
+    uco_amount = validation_nodes_reward()
+
+    nodes =
+      P2P.authorized_nodes()
+      |> Enum.map(fn %Node{reward_address: reward_address} ->
+        {reward_address, uco_amount}
+      end)
+
+    network_pool_balance =
+      SharedSecrets.get_network_pool_address()
+      |> Account.get_balance()
+      |> Map.get(:token)
+      |> Map.to_list()
+      |> Enum.sort(fn {_, qty1}, {_, qty2} -> qty1 < qty2 end)
+
+    do_get_transfers(nodes, network_pool_balance, [])
+  end
+
+  defp do_get_transfers([node | rest], network_pool_balance, acc) do
+    {address, amount} = node
+
+    {transfers, network_pool_balance} =
+      get_node_transfers(address, network_pool_balance, amount, [])
+
+    do_get_transfers(rest, network_pool_balance, Enum.concat(acc, transfers))
   end
 
+  defp do_get_transfers([], _, acc), do: acc
+
+  defp get_node_transfers(reward_address, [token | rest], amount, acc) when amount > 0 do
+    {{token_address, token_id}, token_amount} = token
+
+    if amount >= token_amount do
+      transfer = %Transfer{
+        amount: token_amount,
+        to: reward_address,
+        token: token_address,
+        token_id: token_id
+      }
+
+      amount = amount - token_amount
+
+      get_node_transfers(reward_address, rest, amount, [transfer | acc])
+    else
+      transfer = %Transfer{
+        amount: amount,
+        to: reward_address,
+        token: token_address,
+        token_id: token_id
+      }
+
+      token = {{token_address, token_id}, token_amount - amount}
+
+      get_node_transfers(reward_address, [token | rest], 0, [transfer | acc])
+    end
+  end
+
+  defp get_node_transfers(_, network_pool_balance, 0, acc), do: {acc, network_pool_balance}
+
+  defp get_node_transfers(_, [], _, acc), do: {acc, []}
+
   @doc """
   Returns the last date of the rewards scheduling from the network pool
   """

lib/archethic/reward/scheduler.ex

@@ -88,37 +88,56 @@ defmodule Archethic.Reward.Scheduler do
   def handle_info(:mint_rewards, state = %{interval: interval}) do
     timer = schedule(interval)
 
+    tx_address = Reward.next_address()
+
+    if Reward.initiator?(tx_address) do
+      mint_node_rewards()
+    else
+      DetectNodeResponsiveness.start_link(tx_address, fn count ->
+        if Reward.initiator?(tx_address, count) do
+          Logger.debug("Mint reward creation...attempt #{count}",
+            transaction_address: Base.encode16(tx_address)
+          )
+
+          mint_node_rewards()
+        end
+      end)
+    end
+
+    {:noreply, Map.put(state, :timer, timer), :hibernate}
+  end
+
+  def handle_info({:new_transaction, _address, :mint_rewards, _timestamp}, state) do
+    send_node_rewards()
+    {:noreply, state, :hibernate}
+  end
+
+  def handle_info({:new_transaction, _address, _type, _timestamp}, state) do
+    {:noreply, state, :hibernate}
+  end
+
+  defp mint_node_rewards do
     case DB.get_latest_burned_fees() do
       0 ->
         Logger.info("No mint rewards transaction needed")
+        send_node_rewards()
 
       amount ->
         tx = Reward.new_rewards_mint(amount)
 
-        if Reward.initiator?() do
-          Archethic.send_new_transaction(tx)
+        PubSub.register_to_new_transaction_by_address(tx.address)
 
-          Logger.info("New mint rewards transaction sent with #{amount} token",
-            transaction_address: Base.encode16(tx.address)
-          )
-        else
-          DetectNodeResponsiveness.start_link(tx.address, fn count ->
-            if Reward.initiator?(count) do
-              Logger.debug("Mint secret creation...attempt #{count}",
-                transaction_address: Base.encode16(tx.address)
-              )
-
-              Logger.info("New mint rewards transaction sent with #{amount} token",
-                transaction_address: Base.encode16(tx.address)
-              )
-
-              Archethic.send_new_transaction(tx)
-            end
-          end)
-        end
+        Archethic.send_new_transaction(tx)
+
+        Logger.info("New mint rewards transaction sent with #{amount} token",
+          transaction_address: Base.encode16(tx.address)
+        )
     end
+  end
 
-    {:noreply, Map.put(state, :timer, timer), :hibernate}
+  defp send_node_rewards do
+    Reward.new_node_rewards()
+    |> Archethic.send_new_transaction()
   end
 
   def handle_call(:last_date, _, state = %{interval: interval}) do

lib/archethic/transaction_chain.ex

@@ -497,20 +497,30 @@ defmodule Archethic.TransactionChain do
     addresses =
       tx
       |> Transaction.get_movements()
-      |> Enum.map(& &1.to)
+      |> Enum.map(&{&1.to, &1.type})
       |> Enum.concat(recipients)
 
     Task.Supervisor.async_stream_nolink(
       TaskSupervisor,
       addresses,
-      fn to ->
-        case resolve_last_address(to, time) do
-          {:ok, resolved} ->
-            {to, resolved}
-
-          _ ->
-            {to, to}
-        end
+      fn
+        {to, type} ->
+          case resolve_last_address(to, time) do
+            {:ok, resolved} ->
+              {{to, type}, resolved}
+
+            _ ->
+              {{to, type}, to}
+          end
+
+        to ->
+          case resolve_last_address(to, time) do
+            {:ok, resolved} ->
+              {to, resolved}
+
+            _ ->
+              {to, to}
+          end
       end,
       on_timeout: :kill_task
     )

test/archethic/beacon_chain/subset_test.exs

@@ -278,13 +278,6 @@ defmodule Archethic.BeaconChain.SubsetTest do
         fee: 0
       }
 
-      send(
-        pid,
-        {:new_replication_attestation, %ReplicationAttestation{transaction_summary: tx_summary}}
-      )
-
-      me = self()
-
       MockClient
       |> stub(:send_message, fn
         _, %Ping{}, _ ->
@@ -295,6 +288,13 @@ defmodule Archethic.BeaconChain.SubsetTest do
           {:ok, %Ok{}}
       end)
 
+      send(
+        pid,
+        {:new_replication_attestation, %ReplicationAttestation{transaction_summary: tx_summary}}
+      )
+
+      me = self()
+
       MockDB
       |> stub(:write_transaction_at, fn
         %Transaction{

test/archethic/reward/scheduler_test.exs

@@ -46,18 +46,23 @@ defmodule Archethic.Reward.SchedulerTest do
     assert_receive {:trace, ^pid, :receive, :mint_rewards}, 3_000
   end
 
-  test "should send transaction when burning fees > 0" do
+  test "should send mint transaction when burning fees > 0 and node reward transaction" do
     MockDB
     |> stub(:get_latest_burned_fees, fn -> 15_000 end)
 
     me = self()
 
+    assert {:ok, pid} = Scheduler.start_link(interval: "*/1 * * * * *")
+
     MockClient
-    |> stub(:send_message, fn _, %StartMining{transaction: %{type: type}}, _ ->
-      send(me, type)
-    end)
+    |> stub(:send_message, fn
+      _, %StartMining{transaction: %{type: type}}, _ when type == :mint_rewards ->
+        send(pid, {:new_transaction, nil, :mint_rewards, nil})
+        send(me, type)
 
-    assert {:ok, pid} = Scheduler.start_link(interval: "*/1 * * * * *")
+      _, %StartMining{transaction: %{type: type}}, _ when type == :node_rewards ->
+        send(me, type)
+    end)
 
     send(
       pid,
@@ -70,9 +75,10 @@ defmodule Archethic.Reward.SchedulerTest do
     )
 
     assert_receive :mint_rewards, 1_500
+    assert_receive :node_rewards, 1_500
   end
 
-  test "should not send transaction when burning fees = 0" do
+  test "should not send transaction when burning fees = 0 and should send node rewards" do
     MockDB
     |> stub(:get_latest_burned_fees, fn -> 0 end)
 
@@ -95,6 +101,7 @@ defmodule Archethic.Reward.SchedulerTest do
        }}
     )
 
-    refute_receive :mint_rewards, 1_500
+    refute_receive :mint_rewards, 1_200
+    assert_receive :node_rewards, 1_500
   end
 end