feat(sync-service): add WebSocket wake multiplexing

.changeset/calm-shapes-multiplex.md

@@ -0,0 +1,8 @@
+---
+"@core/sync-service": minor
+---
+
+Add an authenticated, active-instance-only WebSocket endpoint for multiplexing
+silent live shape waits. The endpoint coalesces change subscriptions by shape,
+wakes proxies without routing shape data through the socket, and reproduces
+normal empty live responses when Electric's configured deadline expires.

packages/sync-service/README.md

@@ -82,3 +82,40 @@ The Electric app will startup along with the rest of your Elixir app.
 Beyond the required database connection configuration there are a lot of other
 optional configuration parameters. See the [`Electric` docs for more
 information](https://hexdocs.pm/electric/Electric.html).
+
+## Internal live-request multiplexing
+
+Electric exposes an authenticated WebSocket upgrade at
+`GET /v1/shape/multiplex` for trusted proxies that need to park many silent
+live waits without retaining one HTTP request process per wait. This endpoint
+uses the `electric.shape-multiplex.v1` WebSocket subprotocol. It is only
+available on the active Electric instance; read-only instances reject the
+upgrade.
+
+The client sends JSON text frames to add or remove logical waits:
+
+```json
+{"type":"watch","id":"request-1","handle":"...","offset":"12_0","cursor":"123"}
+{"type":"unwatch","id":"request-1"}
+```
+
+`cursor` is the raw value of the previous `electric-cursor` header, or `null`
+when there is no previous value. Electric may acknowledge an armed wait with
+`{"type":"ready","id":"request-1"}`. It then sends one terminal frame:
+
+```json
+{"type":"wake","id":"request-1","reason":"changes"}
+{"type":"wake","id":"request-1","reason":"rotation"}
+{"type":"no_change","id":"request-1","response":{"status":200,"headers":{},"body":[]}}
+{"type":"error","id":"request-1","code":"...","message":"...","retryable":true}
+```
+
+Wake frames deliberately carry no shape rows. The proxy must issue the normal
+shape HTTP request after a wake. A `no_change` frame contains the lowercase
+HTTP response headers and JSON body that the proxy should return to the
+unchanged shape client. The server uses its configured long-poll timeout and
+removes every wait after a terminal frame.
+
+Embedded deployments can call `Electric.Plug.ShapeMultiplexPlug` directly. An
+optional `:availability_guard` zero-arity function can enforce tenant ownership;
+it must return `:ok` while the socket is valid and is rechecked periodically.

packages/sync-service/lib/electric/plug/router.ex

@@ -35,6 +35,10 @@ defmodule Electric.Plug.Router do
     to: PassAssignToOptsPlug,
     init_opts: [plug: Electric.Plug.ServeShapePlug, assign_key: :config]
 
+  get "/v1/shape/multiplex",
+    to: PassAssignToOptsPlug,
+    init_opts: [plug: Electric.Plug.ShapeMultiplexPlug, assign_key: :config]
+
   post "/v1/shape",
     to: PassAssignToOptsPlug,
     init_opts: [plug: Electric.Plug.ServeShapePlug, assign_key: :config]
@@ -55,7 +59,8 @@ defmodule Electric.Plug.Router do
   # OPTIONS requests should not be authenticated
   def authenticate(%Plug.Conn{method: "OPTIONS"} = conn, _opts), do: conn
 
-  def authenticate(%Plug.Conn{request_path: "/v1/shape"} = conn, _opts) do
+  def authenticate(%Plug.Conn{request_path: request_path} = conn, _opts)
+      when request_path in ["/v1/shape", "/v1/shape/multiplex"] do
     api_secret = conn.assigns.config[:secret]
 
     if is_nil(api_secret) do

packages/sync-service/lib/electric/plug/shape_multiplex_plug.ex

@@ -0,0 +1,121 @@
+defmodule Electric.Plug.ShapeMultiplexPlug do
+  @moduledoc """
+  Upgrades an authenticated request to the internal shape live-wait multiplexer.
+
+  The standalone router exposes this Plug at `GET /v1/shape/multiplex`. Embedded
+  deployments may invoke it directly with the normal Electric API options plus:
+
+    * `:availability_guard` — optional zero-arity function returning `:ok` while
+      this process owns the tenant and `{:error, reason}` otherwise. It is
+      checked before upgrade and periodically for established sockets.
+    * `:subprotocol` — selected WebSocket subprotocol; defaults to
+      `electric.shape-multiplex.v1`.
+
+  The Plug also requires the underlying Electric stack to be the active
+  instance. Authentication remains the responsibility of the enclosing router.
+  """
+
+  @behaviour Plug
+
+  alias Electric.Shapes.Api.Multiplex
+  alias Electric.Shapes.Api.Multiplex.Source
+  alias Electric.Shapes.Api.Multiplex.WebSocket
+
+  @max_frame_size 65_536
+
+  @impl Plug
+  def init(opts), do: opts
+
+  @impl Plug
+  def call(conn, opts) do
+    api = fetch_opt!(opts, :api)
+    source = Access.get(opts, :multiplex_source, Source)
+    source_opts = Access.get(opts, :multiplex_source_opts)
+    availability_guard = Access.get(opts, :availability_guard)
+    subprotocol = Access.get(opts, :subprotocol, Multiplex.protocol())
+
+    cond do
+      not Multiplex.available?(api, source, source_opts, availability_guard) ->
+        error_response(
+          conn,
+          503,
+          "inactive_instance",
+          "Multiplexing is only available on the active Electric instance",
+          true,
+          retry_after: 1
+        )
+
+      not offered_subprotocol?(conn, subprotocol) ->
+        error_response(
+          conn,
+          400,
+          "unsupported_subprotocol",
+          "WebSocket subprotocol #{subprotocol} is required",
+          false
+        )
+
+      true ->
+        case WebSockAdapter.UpgradeValidation.validate_upgrade(conn) do
+          :ok ->
+            socket_opts = %{
+              api: api,
+              availability_guard: availability_guard,
+              multiplex_source: source,
+              multiplex_source_opts: source_opts,
+              multiplex_status_check_interval:
+                Access.get(opts, :multiplex_status_check_interval, 5_000)
+            }
+
+            conn
+            |> Plug.Conn.put_resp_header("sec-websocket-protocol", subprotocol)
+            |> WebSockAdapter.upgrade(WebSocket, socket_opts,
+              early_validate_upgrade: false,
+              max_frame_size: Access.get(opts, :multiplex_max_frame_size, @max_frame_size),
+              timeout: max(60_000, api.long_poll_timeout * 3)
+            )
+
+          {:error, reason} ->
+            error_response(conn, 400, "invalid_upgrade", reason, false)
+        end
+    end
+  end
+
+  defp offered_subprotocol?(conn, subprotocol) when is_binary(subprotocol) do
+    conn
+    |> Plug.Conn.get_req_header("sec-websocket-protocol")
+    |> Enum.flat_map(&Plug.Conn.Utils.list/1)
+    |> Enum.any?(&(&1 == subprotocol))
+  end
+
+  defp offered_subprotocol?(_conn, _subprotocol), do: false
+
+  defp error_response(conn, status, code, message, retryable, opts \\ []) do
+    conn =
+      conn
+      |> Plug.Conn.put_resp_content_type("application/json")
+      |> Plug.Conn.put_resp_header("cache-control", "no-store")
+      |> Plug.Conn.put_resp_header("surrogate-control", "no-store")
+
+    conn =
+      case Keyword.fetch(opts, :retry_after) do
+        {:ok, seconds} ->
+          Plug.Conn.put_resp_header(conn, "retry-after", Integer.to_string(seconds))
+
+        :error ->
+          conn
+      end
+
+    Plug.Conn.send_resp(
+      conn,
+      status,
+      Jason.encode!(%{code: code, message: message, retryable: retryable})
+    )
+  end
+
+  defp fetch_opt!(opts, key) do
+    case Access.fetch(opts, key) do
+      {:ok, value} -> value
+      :error -> raise KeyError, key: key, term: opts
+    end
+  end
+end

packages/sync-service/lib/electric/shape_cache.ex

@@ -585,7 +585,8 @@ defmodule Electric.ShapeCache do
 
   @spec fetch_latest_offset(stack_id(), shape_handle(), keyword()) ::
           {:ok, LogOffset.t()} | :error
-  defp fetch_latest_offset(stack_id, shape_handle, opts \\ []) do
+  @doc false
+  def fetch_latest_offset(stack_id, shape_handle, opts \\ []) do
     storage =
       Storage.for_shape(shape_handle, Storage.for_stack(stack_id, read_only?: opts[:read_only?]))
 

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

@@ -123,6 +123,16 @@ defmodule Electric.Shapes do
     ShapeCache.has_shape?(shape_handle, stack_id)
   end
 
+  @doc false
+  @spec fetch_latest_offset(stack_id(), shape_handle()) :: {:ok, LogOffset.t()} | :error
+  def fetch_latest_offset(stack_id, shape_handle) do
+    if ShapeCache.has_shape?(shape_handle, stack_id) do
+      ShapeCache.fetch_latest_offset(stack_id, shape_handle)
+    else
+      :error
+    end
+  end
+
   @doc """
   Remove and clean up all data (meta data and shape log + snapshot) associated with
   the given shape handle

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

@@ -0,0 +1,124 @@
+defmodule Electric.Shapes.Api.Multiplex do
+  @moduledoc """
+  Protocol helpers for Electric's internal live-request multiplexer.
+
+  The WebSocket subprotocol is `electric.shape-multiplex.v1`. A client adds and
+  removes logical live requests with JSON text frames:
+
+      {"type":"watch","id":"request-1","handle":"...","offset":"12_0","cursor":"123"}
+      {"type":"unwatch","id":"request-1"}
+
+  `cursor` is the unwrapped value of the previous `electric-cursor` response
+  header. It may be `null` for a request without a previous cursor.
+
+  An accepted watch receives an optional `ready` frame, followed by exactly one
+  terminal `wake` or `no_change` frame unless the client unwatches it. Wake
+  frames never contain shape data; the caller fetches that data through the
+  normal shape HTTP endpoint. A `no_change` frame contains the status, headers,
+  and JSON body needed to reproduce the normal empty live HTTP response.
+  """
+
+  alias Electric.Postgres.Lsn
+  alias Electric.Replication.LogOffset
+  alias Electric.Shapes.Api
+  alias Electric.Shapes.Api.Params
+  alias Electric.Shapes.Api.Response
+
+  @protocol "electric.shape-multiplex.v1"
+
+  @type watch_id :: binary()
+  @type wake_reason :: :changes | :rotation
+  @type server_frame ::
+          %{required(:type) => binary(), required(:id) => watch_id()}
+          | %{
+              required(:type) => binary(),
+              required(:code) => binary(),
+              required(:message) => binary(),
+              required(:retryable) => boolean(),
+              optional(:id) => watch_id()
+            }
+
+  @spec protocol() :: binary()
+  def protocol, do: @protocol
+
+  @doc false
+  def available?(%Api{} = api, source, source_opts, availability_guard) do
+    source.active?(api, source_opts) and guard_available?(availability_guard)
+  rescue
+    _ -> false
+  catch
+    _, _ -> false
+  end
+
+  @spec ready_frame(watch_id()) :: server_frame()
+  def ready_frame(id), do: %{type: "ready", id: id}
+
+  @spec wake_frame(watch_id(), wake_reason()) :: server_frame()
+  def wake_frame(id, reason) when reason in [:changes, :rotation] do
+    %{type: "wake", id: id, reason: Atom.to_string(reason)}
+  end
+
+  @spec error_frame(nil | watch_id(), binary(), binary(), boolean()) :: server_frame()
+  def error_frame(id, code, message, retryable) do
+    %{type: "error", code: code, message: message, retryable: retryable}
+    |> maybe_put_id(id)
+  end
+
+  @doc """
+  Builds the transport-neutral equivalent of an empty live shape response.
+
+  Header generation deliberately goes through `Api.Response` so cache policy,
+  ETag format, and cursor advancement remain identical to the HTTP path.
+  """
+  @spec no_change_frame(
+          watch_id(),
+          Api.t(),
+          Electric.shape_handle(),
+          LogOffset.t(),
+          binary() | nil
+        ) :: server_frame()
+  def no_change_frame(id, %Api{} = api, handle, %LogOffset{} = offset, cursor) do
+    global_last_seen_lsn =
+      case Electric.LsnTracker.get_last_processed_lsn(api.stack_id) do
+        nil -> offset.tx_offset
+        lsn -> Lsn.to_integer(lsn)
+      end
+
+    body = [
+      %{
+        headers: %{
+          control: "up-to-date",
+          global_last_seen_lsn: to_string(global_last_seen_lsn)
+        }
+      }
+    ]
+
+    params = %Params{handle: handle, offset: offset, live: true}
+
+    response = %Response{
+      api: api,
+      handle: handle,
+      offset: offset,
+      params: params,
+      status: 200,
+      up_to_date: true,
+      no_changes: true,
+      body: body,
+      finalized?: true
+    }
+
+    headers = Response.client_headers(response, %{"cursor" => cursor})
+
+    %{
+      type: "no_change",
+      id: id,
+      response: %{status: 200, headers: headers, body: body}
+    }
+  end
+
+  defp maybe_put_id(frame, nil), do: frame
+  defp maybe_put_id(frame, id), do: Map.put(frame, :id, id)
+
+  defp guard_available?(nil), do: true
+  defp guard_available?(guard) when is_function(guard, 0), do: guard.() == :ok
+end

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

@@ -0,0 +1,39 @@
+defmodule Electric.Shapes.Api.Multiplex.Source do
+  @moduledoc false
+
+  alias Electric.Shapes
+  alias Electric.Shapes.Api
+
+  @callback active?(Api.t(), term()) :: boolean()
+  @callback lookup(Api.t(), Electric.shape_handle(), term()) ::
+              {:ok, Electric.Replication.LogOffset.t()} | :not_found
+  @callback subscribe(Api.t(), Electric.shape_handle(), reference(), term()) :: :ok
+  @callback unsubscribe(Api.t(), Electric.shape_handle(), term()) :: :ok
+
+  def active?(%Api{stack_id: stack_id}, _opts) do
+    Electric.StatusMonitor.service_status(stack_id) == :active
+  end
+
+  def lookup(%Api{stack_id: stack_id}, handle, _opts) do
+    with {:ok, latest_offset} <- Shapes.fetch_latest_offset(stack_id, handle) do
+      {:ok, latest_offset}
+    else
+      _ -> :not_found
+    end
+  end
+
+  def subscribe(%Api{stack_id: stack_id}, handle, ref, _opts) do
+    ^ref = Electric.StackSupervisor.subscribe_to_shape_events(stack_id, handle, ref)
+    :ok
+  end
+
+  def unsubscribe(%Api{stack_id: stack_id}, handle, _opts) do
+    registry = Electric.StackSupervisor.registry_name(stack_id)
+
+    if GenServer.whereis(registry) != nil do
+      Registry.unregister(registry, handle)
+    end
+
+    :ok
+  end
+end