feat: support for Server-Sent Events (SSE) (#2776)

This is a follow up PR on
#2546 and
#2544. It solves a bug
related with 409s (must refetch) in SSE mode and it replaces the
EventSource browser API by the
[fetch-event-source](https://github.com/Azure/fetch-event-source)
library. I refactored the `ShapeStream.#start` method which was becoming
very big and complex. To this end, i split the logic into helper methods
that handle the different parts that need to happen (building the shape
URL, making the request, parsing the response headers, handling the
response body, etc.).

I had to patch the
[fetch-event-source](https://github.com/Azure/fetch-event-source)
library because it relies on browser-specific features such as
`document` and `window` (cf.
Azure/fetch-event-source#41). But we want our
client to also work in server-side JS environments.

I also had to patch the `fetch-event-source` library because it does not
abort the fetch when you pass an already aborted signal. A complete
description of the bug and the fix can be found here:
Azure/fetch-event-source#98.

Author: kevin-dp

package.json

@@ -20,5 +20,10 @@
   },
   "devDependencies": {
     "glob": "^10.3.10"
+  },
+  "pnpm": {
+    "patchedDependencies": {
+      "@microsoft/fetch-event-source": "patches/@microsoft__fetch-event-source.patch"
+    }
   }
 }

packages/sync-service/lib/electric/shapes/api.ex

@@ -317,8 +317,18 @@ defmodule Electric.Shapes.Api do
 
     # TODO: discuss returning a 307 redirect rather than a 409, the client
     # will have to detect this and throw out old data
+
+    # In SSE mode we send the must refetch object as an event
+    # instead of a singleton array containing that object
+    must_refetch =
+      if request.params.experimental_live_sse do
+        hd(@must_refetch)
+      else
+        @must_refetch
+      end
+
     {:error,
-     Response.error(request, @must_refetch,
+     Response.error(request, must_refetch,
        handle: active_shape_handle,
        status: 409
      )}
@@ -610,10 +620,9 @@ defmodule Electric.Shapes.Api do
       last_message_time: last_message_time,
       request:
         %{
-          api:
-            %{
-              keepalive_interval: keepalive_interval
-            } = api,
+          api: %{
+            keepalive_interval: keepalive_interval
+          },
           handle: shape_handle,
           new_changes_ref: ref
         } = request,
@@ -664,7 +673,7 @@ defmodule Electric.Shapes.Api do
 
       {^ref, :shape_rotation} ->
         must_refetch = %{headers: %{control: "must-refetch"}}
-        message = encode_message(api, must_refetch)
+        message = encode_message(request, must_refetch)
 
         {message, %{state | mode: :done}}
 
@@ -793,11 +802,19 @@ defmodule Electric.Shapes.Api do
     encode(api, :log, stream)
   end
 
-  @spec encode_message(Api.t() | Request.t(), term()) :: Enum.t()
-  def encode_message(%Api{} = api, message) do
+  # Error messages are encoded normally, even when using SSE
+  # because they are returned on the original fetch request
+  # with a status code that is not 2xx.
+  @spec encode_error_message(Api.t() | Request.t(), term()) :: Enum.t()
+  def encode_error_message(%Api{} = api, message) do
+    encode(api, :message, message)
+  end
+
+  def encode_error_message(%Request{api: api}, message) do
     encode(api, :message, message)
   end
 
+  @spec encode_message(Request.t(), term()) :: Enum.t()
   def encode_message(
         %Request{api: api, params: %{live: true, experimental_live_sse: true}},
         message

packages/sync-service/lib/electric/shapes/api/response.ex

@@ -86,7 +86,7 @@ defmodule Electric.Shapes.Api.Response do
         message
       end
 
-    Api.encode_message(api_or_request, body)
+    Api.encode_error_message(api_or_request, body)
   end
 
   @spec send(Plug.Conn.t(), t()) :: Plug.Conn.t()

packages/typescript-client/package.json

@@ -6,7 +6,9 @@
   "bugs": {
     "url": "https://github.com/electric-sql/electric/issues"
   },
-  "dependencies": {},
+  "dependencies": {
+    "@microsoft/fetch-event-source": "^2.0.1"
+  },
   "devDependencies": {
     "@types/pg": "^8.11.6",
     "@types/uuid": "^10.0.0",