fix(explorer): add real time updates (#497)

Co-authored-by: Esteban Dimitroff Hodi <[email protected]>
Co-authored-by: Julian Arce <[email protected]>
Co-authored-by: Urix <[email protected]>
Co-authored-by: Mauro Toscano <[email protected]>

Author: 5 people

explorer/lib/explorer/models/batches.ex

@@ -125,8 +125,13 @@ defmodule Batches do
       nil ->
         "New Batch, inserting to DB:" |> IO.puts()
         case Explorer.Repo.insert(changeset) do
-          {:ok, _} -> "Batch inserted successfully" |> IO.puts()
-          {:error, changeset} -> "Batch insert failed #{changeset}" |> IO.puts()
+          {:ok, _} ->
+            "Batch inserted successfully" |> IO.puts()
+            {:ok, :empty}
+
+          {:error, changeset} ->
+            "Batch insert failed #{changeset}" |> IO.puts()
+            {:error, changeset}
         end
       existing_batch ->
         try do
@@ -143,8 +148,13 @@ defmodule Batches do
             "Batch values have changed, updating in DB" |> IO.puts()
             updated_changeset = Ecto.Changeset.change(existing_batch, changeset.changes)
             case Explorer.Repo.update(updated_changeset) do
-              {:ok, _} -> "Batch updated successfully" |> IO.puts()
-              {:error, changeset} -> "Batch update failed #{changeset}" |> IO.puts()
+              {:ok, _} ->
+                "Batch updated successfully" |> IO.puts()
+                {:ok, :empty}
+
+              {:error, changeset} ->
+                "Batch update failed #{changeset}" |> IO.puts()
+                {:error, changeset}
             end
           end
         rescue

explorer/lib/explorer/periodically.ex

@@ -1,4 +1,5 @@
 defmodule Explorer.Periodically do
+  alias Phoenix.PubSub
   use GenServer
 
   def start_link(_) do
@@ -44,6 +45,7 @@ defmodule Explorer.Periodically do
     rescue
       error -> IO.puts("An error occurred during batch processing:\n#{inspect(error)}")
     end
+
     IO.inspect("Done processing from block #{fromBlock} to block #{toBlock}")
   end
 
@@ -62,6 +64,16 @@ defmodule Explorer.Periodically do
           |> Utils.extract_amount_of_proofs
           |> Batches.generate_changeset
           |> Batches.insert_or_update
+          |> case do
+            {:ok, _} ->
+              IO.puts("Broadcasting update_views")
+              PubSub.broadcast(Explorer.PubSub, "update_views", %{})
+            {:error, error} ->
+              IO.puts("Some error in DB operation, not broadcasting update_views")
+              IO.inspect(error)
+            nil -> nil #no changes in DB
+
+          end
 
         "Done processing batch: #{batch.merkle_root}" |> IO.inspect
         Mutex.release(BatchMutex, lock)

explorer/lib/explorer_web/live/pages/batch/index.ex

@@ -5,6 +5,8 @@ defmodule ExplorerWeb.Batch.Index do
   def mount(params, _, socket) do
     merkle_root = params["merkle_root"]
 
+    Phoenix.PubSub.subscribe(Explorer.PubSub, "update_views")
+
     if merkle_root == nil do
       {
         :empty,
@@ -18,7 +20,6 @@ defmodule ExplorerWeb.Batch.Index do
         batch -> batch
       end
 
-
     {
       :ok,
       assign(socket,
@@ -32,5 +33,17 @@ defmodule ExplorerWeb.Batch.Index do
       {:ok, assign(socket, merkle_root: :empty, newBatchInfo: :empty, batchWasResponded: :empty)}
   end
 
+  def handle_info(_, socket) do
+    IO.puts("Received batch update for #{socket.assigns.merkle_root} from PubSub")
+
+    {
+      :noreply,
+      assign(
+        socket,
+        current_batch: Batches.get_batch(%{merkle_root: socket.assigns.merkle_root})
+      )
+    }
+  end
+
   embed_templates "*"
 end

explorer/lib/explorer_web/live/pages/batches/index.ex

@@ -1,17 +1,30 @@
 defmodule ExplorerWeb.Batches.Index do
+  alias Phoenix.PubSub
   require Logger
   use ExplorerWeb, :live_view
 
+  @page_size 12
+
   def mount(params, _, socket) do
     current_page = get_current_page(params)
 
-    page_size = 12
+    batches = Batches.get_latest_batches(%{amount: @page_size * current_page})
 
-    batches = Batches.get_latest_batches(%{amount: page_size * current_page})
+    PubSub.subscribe(Explorer.PubSub, "update_views")
 
     {:ok, assign(socket, current_page: current_page, batches: batches, page_title: "Batches")}
   end
 
+  def handle_info(_, socket) do
+    IO.puts("Received update for batches from PubSub")
+
+    current_page = socket.assigns.current_page
+
+    batches = Batches.get_latest_batches(%{amount: @page_size * current_page})
+
+    {:noreply, assign(socket, batches: batches)}
+  end
+
   def get_current_page(params) do
     case params |> Map.get("page") do
       nil ->

explorer/lib/explorer_web/live/pages/home/index.ex

@@ -19,6 +19,30 @@ defmodule ExplorerWeb.Home.Index do
     end
   end
 
+  def handle_info(_, socket) do
+    IO.puts("Received update for home from PubSub")
+
+    verified_batches = Batches.get_amount_of_verified_batches()
+
+    operators_registered = get_operators_registered()
+
+    latest_batches =
+      Batches.get_latest_batches(%{amount: 5})
+      # extract only the merkle root
+      |> Enum.map(fn %Batches{merkle_root: merkle_root} -> merkle_root end)
+
+    verified_proofs = Batches.get_amount_of_verified_proofs()
+
+    {:noreply,
+     assign(
+       socket,
+       verified_batches: verified_batches,
+       operators_registered: operators_registered,
+       latest_batches: latest_batches,
+       verified_proofs: verified_proofs
+     )}
+  end
+
   def mount(_, _, socket) do
     verified_batches = Batches.get_amount_of_verified_batches()
 
@@ -31,6 +55,8 @@ defmodule ExplorerWeb.Home.Index do
 
     verified_proofs = Batches.get_amount_of_verified_proofs()
 
+    Phoenix.PubSub.subscribe(Explorer.PubSub, "update_views")
+
     {:ok,
      assign(socket,
        verified_batches: verified_batches,