WIP feat(client): experimental support for streaming SSE in live mode (#2546)

Co-authored-by: Kevin <kevin@electric-sql.com>

Author: samwillis

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",