Fixed multiple pub sub causing crash due to new replication attestati…

…ons (#432)

lib/archethic/beacon_chain/subset.ex

@@ -193,11 +193,13 @@ defmodule Archethic.BeaconChain.Subset do
     end
   end
 
-  defp notify_subscribed_nodes(nodes, attestation) do
+  defp notify_subscribed_nodes(nodes, %ReplicationAttestation{
+         transaction_summary: tx_summary
+       }) do
     nodes
     |> P2P.get_nodes_info()
     |> Enum.reject(&(&1.first_public_key == Crypto.first_node_public_key()))
-    |> P2P.broadcast_message(attestation)
+    |> P2P.broadcast_message(tx_summary)
   end
 
   defp handle_slot(

lib/archethic/p2p/message.ex

@@ -1304,7 +1304,17 @@ defmodule Archethic.P2P.Message do
   end
 
   def process(%BeaconUpdate{transaction_attestations: transaction_attestations}) do
-    Enum.each(transaction_attestations, &process/1)
+    Enum.each(transaction_attestations, fn %ReplicationAttestation{
+                                             transaction_summary: tx_summary
+                                           } ->
+      process(tx_summary)
+    end)
+
+    %Ok{}
+  end
+
+  def process(tx_summary = %TransactionSummary{}) do
+    PubSub.notify_transaction_attestation(tx_summary)
 
     %Ok{}
   end

lib/archethic/pub_sub.ex

@@ -16,6 +16,8 @@ defmodule Archethic.PubSub do
 
   alias Archethic.TransactionChain.Transaction
 
+  alias Archethic.TransactionChain.TransactionSummary
+
   @doc """
   Notify the registered processes than a new transaction has been validated
   """
@@ -90,6 +92,17 @@ defmodule Archethic.PubSub do
     )
   end
 
+  @doc """
+  Notify a new transaction  attestation for beacon explorer
+  """
+  @spec notify_transaction_attestation(TransactionSummary.t()) :: :ok
+  def notify_transaction_attestation(attestation = %TransactionSummary{}) do
+    dispatch(
+      :new_transaction_attestation,
+      {:new_transaction_attestation, attestation}
+    )
+  end
+
   @doc """
   Register a process to a new transaction publication by type
   """
@@ -177,6 +190,13 @@ defmodule Archethic.PubSub do
     Registry.register(PubSubRegistry, :new_replication_attestation, [])
   end
 
+  @doc """
+  Register to new transaction attestations for beacon explorer
+  """
+  def register_to_new_transaction_attestations do
+    Registry.register(PubSubRegistry, :new_transaction_attestation, [])
+  end
+
   defp dispatch(topic, message) do
     Registry.dispatch(PubSubRegistry, topic, fn entries ->
       for {pid, _} <- entries, do: send(pid, message)

lib/archethic_web/live/chains/beacon_live.ex

@@ -3,7 +3,6 @@ defmodule ArchethicWeb.BeaconChainLive do
   use ArchethicWeb, :live_view
 
   alias Archethic.BeaconChain
-  alias Archethic.BeaconChain.ReplicationAttestation
   alias Archethic.BeaconChain.Slot
   alias Archethic.BeaconChain.SummaryTimer
 
@@ -19,7 +18,7 @@ defmodule ArchethicWeb.BeaconChainLive do
   alias Archethic.TransactionChain
   alias Archethic.TransactionChain.Transaction
   alias Archethic.TransactionChain.TransactionData
-
+  alias Archethic.TransactionChain.TransactionSummary
   alias ArchethicWeb.ExplorerView
   alias ArchethicWeb.TransactionCache
 
@@ -33,8 +32,7 @@ defmodule ArchethicWeb.BeaconChainLive do
     if connected?(socket) do
       PubSub.register_to_next_summary_time()
       PubSub.register_to_current_epoch_of_slot_time()
-      PubSub.register_to_new_replication_attestations()
-
+      PubSub.register_to_new_transaction_attestations()
       # register for client to able to get the current added transaction to the beacon pool
       BeaconChain.register_to_beacon_pool_updates()
     end
@@ -141,7 +139,7 @@ defmodule ArchethicWeb.BeaconChainLive do
   end
 
   def handle_info(
-        {:new_replication_attestation, %ReplicationAttestation{transaction_summary: tx_summary}},
+        {:new_transaction_attestation, tx_summary = %TransactionSummary{}},
         socket = %{
           assigns:
             assigns = %{