feat: sse implementation for events beacon API

config/config.exs

@@ -52,3 +52,13 @@ if System.get_env("RUSTLER_SKIP_COMPILE") do
   config :lambda_ethereum_consensus, Snappy, skip_compilation?: true
   config :lambda_ethereum_consensus, Ssz, skip_compilation?: true
 end
+
+config :sse,
+  keep_alive: {:system, "SSE_KEEP_ALIVE_IN_MS", 1000}
+
+config :event_bus,
+  topics: [:finalized_checkpoint]
+
+config :mime, :types, %{
+  "text/event-stream" => ["sse"]
+}

lib/beacon_api/controllers/v1/events.ex

@@ -0,0 +1,35 @@
+defmodule BeaconApi.V1.Events do
+  use BeaconApi, :controller
+
+  alias BeaconApi.ApiSpec
+  alias BeaconApi.Helpers
+  alias SSE.Chunk
+
+  require Logger
+
+  @topic :finalized_checkpoint
+
+  def open_api_operation(:subscribe),
+  do: ApiSpec.spec().paths["/eth/v1/events"].get
+
+  @spec subscribe(Plug.Conn.t(), any) :: Plug.Conn.t()
+  def subscribe(conn, _params) do
+    Logger.info("Subscribing to finalized checkpoint events")
+    _finalized_checkpoint = get_current_finalized_checkpoint()
+
+    Logger.info("Sending SSE stream")
+    chunk = %Chunk{data: %{} |> Jason.encode!()}
+
+    Logger.info(chunk |> inspect())
+    conn
+    |> SSE.stream({[@topic], chunk})
+  end
+
+  defp get_current_finalized_checkpoint() do
+    x = Helpers.finalized_checkpoint()
+
+    Logger.info(x |> inspect())
+
+    x
+  end
+end

lib/beacon_api/helpers.ex

@@ -161,6 +161,9 @@ defmodule BeaconApi.Helpers do
     end
   end
 
+  @spec finalized_checkpoint() :: Types.Checkpoint.t()
+  def finalized_checkpoint(), do: ForkChoice.get_finalized_checkpoint()
+
   @spec get_state_root(Types.root()) :: Types.root() | nil
   def get_state_root(root) do
     with %{} = block <- Blocks.get_block(root) do

lib/beacon_api/router.ex

@@ -2,7 +2,7 @@ defmodule BeaconApi.Router do
   use BeaconApi, :router
 
   pipeline :api do
-    plug(:accepts, ["json"])
+    plug(:accepts, ["json", "sse"])
     plug(OpenApiSpex.Plug.PutApiSpec, module: BeaconApi.ApiSpec)
   end
 
@@ -22,6 +22,10 @@ defmodule BeaconApi.Router do
       get("/identity", NodeController, :identity)
       get("/version", NodeController, :version)
     end
+
+    scope "/events" do
+      get("/", Events, :subscribe)
+    end
   end
 
   # Ethereum API Version 2

mix.exs

@@ -64,7 +64,10 @@ defmodule LambdaEthereumConsensus.MixProject do
       {:sentry, "~> 10.8.0"},
       {:prom_ex, "~> 1.11.0"},
       {:flama, git: "https://github.com/lambdaclass/ht1223_tracer"},
-      {:uuid, "~> 1.1"}
+      {:uuid, "~> 1.1"},
+      # TODO: We might want to use phoenix_pubsub instead and do our implementation of SSE.
+      {:sse, "~> 0.4"},
+      {:event_bus, ">= 1.6.0"}
     ]
   end
 

mix.lock

@@ -19,6 +19,7 @@
   "elixir_make": {:hex, :elixir_make, "0.8.4", "4960a03ce79081dee8fe119d80ad372c4e7badb84c493cc75983f9d3bc8bde0f", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:certifi, "~> 2.0", [hex: :certifi, repo: "hexpm", optional: true]}], "hexpm", "6e7f1d619b5f61dfabd0a20aa268e575572b542ac31723293a4c1a567d5ef040"},
   "erlex": {:hex, :erlex, "0.2.7", "810e8725f96ab74d17aac676e748627a07bc87eb950d2b83acd29dc047a30595", [:mix], [], "hexpm", "3ed95f79d1a844c3f6bf0cea61e0d5612a42ce56da9c03f01df538685365efb0"},
   "escape": {:hex, :escape, "0.1.0", "548edab75e6e6938b1e199ef59cb8e504bcfd3bcf83471d4ae9a3c7a7a3c7d45", [:mix], [], "hexpm", "a5d8e92db4677155df54bc1306d401b5233875d570d474201db03cb3047491cd"},
+  "event_bus": {:hex, :event_bus, "1.7.0", "29a36fc09e8c4463c82206b6a300fa1d61cf4baf9a7b4e7cf0c3efb99c73998e", [:mix], [], "hexpm", "e556470f49f53060a0696c4bad81341252685011afc69eda25032c8a3a86eb2e"},
   "ex2ms": {:hex, :ex2ms, "1.7.0", "45b9f523d0b777667ded60070d82d871a37e294f0b6c5b8eca86771f00f82ee1", [:mix], [], "hexpm", "2589eee51f81f1b1caa6d08c990b1ad409215fe6f64c73f73c67d36ed10be827"},
   "exleveldb": {:hex, :exleveldb, "0.14.0", "8e9353bbce38482d6971d254c6b98ceb50f3f179c94732b5d17db1be426fca18", [:mix], [{:eleveldb, "~> 2.2.20", [hex: :eleveldb, repo: "hexpm", optional: false]}], "hexpm", "803cd3b4c826a1e17e7e28f6afe224837a743b475e1a48336f186af3dd8636ad"},
   "expo": {:hex, :expo, "0.5.2", "beba786aab8e3c5431813d7a44b828e7b922bfa431d6bfbada0904535342efe2", [:mix], [], "hexpm", "8c9bfa06ca017c9cb4020fabe980bc7fdb1aaec059fd004c2ab3bff03b1c599c"},
@@ -68,6 +69,7 @@
   "sentry": {:hex, :sentry, "10.8.1", "aa45309785e1521416225adb16e0b4d8b957578804527f3c7babb6fefbc5e456", [:mix], [{:hackney, "~> 1.8", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:nimble_options, "~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_ownership, "~> 0.3.0 or ~> 1.0", [hex: :nimble_ownership, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.6", [hex: :phoenix, repo: "hexpm", optional: true]}, {:phoenix_live_view, "~> 0.20 or ~> 1.0", [hex: :phoenix_live_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.6", [hex: :plug, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "495b3cdadad90ba72eef973aa3dec39b3b8b2a362fe87e2f4ef32133ac3b4097"},
   "snappyer": {:hex, :snappyer, "1.2.10", "023e9ae00e969b0997208b5de7d3b12bb46ec6bc5411e8dc53e7b3f435b8f0fd", [:rebar3], [], "hexpm", "f55bd9ed147e7163cb3acd1e431a7ff2c9e31ceacbb8308786094fb64551c284"},
   "sourceror": {:hex, :sourceror, "1.5.0", "3e65d5fbb1a8e2864ad6411262c8018fee73474f5789dda12285c82999253d5d", [:mix], [], "hexpm", "4a32b5d189d8453f73278c15712f8731b89e9211e50726b798214b303b51bfc7"},
+  "sse": {:hex, :sse, "0.4.0", "f17affacbc4618bac07590eec7bff849aa27d1f71bb3d41da3fd3cb255d16910", [:mix], [{:event_bus, ">= 1.6.0", [hex: :event_bus, repo: "hexpm", optional: false]}, {:plug, ">= 1.4.5", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "2dfb9923725b9d5292763c3de9b7798713f5771522823e961a250204917d7efb"},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"},
   "statistex": {:hex, :statistex, "1.0.0", "f3dc93f3c0c6c92e5f291704cf62b99b553253d7969e9a5fa713e5481cd858a5", [:mix], [], "hexpm", "ff9d8bee7035028ab4742ff52fc80a2aa35cece833cf5319009b52f1b5a86c27"},
   "stream_data": {:hex, :stream_data, "1.1.2", "05499eaec0443349ff877aaabc6e194e82bda6799b9ce6aaa1aadac15a9fdb4d", [:mix], [], "hexpm", "129558d2c77cbc1eb2f4747acbbea79e181a5da51108457000020a906813a1a9"},