fix: Forward request headers onto sync backend

lib/phoenix/sync/electric/client_adapter.ex

@@ -51,6 +51,8 @@ defmodule Phoenix.Sync.Electric.ClientAdapter do
     defp live?(live), do: live == "true"
 
     defp fetch_upstream(sync_client, conn, request, shape) do
+      request = put_req_headers(request, conn.req_headers)
+
       response =
         case Client.Fetch.request(sync_client.client, request) do
           %Client.Fetch.Response{} = response -> response
@@ -68,18 +70,33 @@ defmodule Phoenix.Sync.Electric.ClientAdapter do
         end
 
       conn
-      |> put_headers(response.headers)
+      |> put_resp_headers(response.headers)
       |> Plug.Conn.send_resp(response.status, body)
     end
 
-    defp put_headers(conn, headers) do
-      headers
-      |> Map.delete("transfer-encoding")
-      |> Enum.reduce(conn, fn {header, values}, conn ->
-        Enum.reduce(values, conn, fn value, conn ->
-          Plug.Conn.put_resp_header(conn, header, value)
+    defp put_req_headers(request, headers) do
+      merged_headers =
+        Enum.reduce(headers, request.headers, fn {header, value}, acc ->
+          Map.update(acc, header, [value], fn existing -> [value | List.wrap(existing)] end)
         end)
-      end)
+        |> expand_headers()
+
+      %{request | headers: merged_headers}
+    end
+
+    defp put_resp_headers(conn, headers) do
+      resp_headers =
+        headers
+        |> Map.delete("transfer-encoding")
+        |> expand_headers()
+
+      Plug.Conn.merge_resp_headers(conn, resp_headers)
+    end
+
+    # turn headers into a list which is more compatible than a map
+    # representation as it preserves multiple values for a header.
+    defp expand_headers(headers) when is_map(headers) do
+      Enum.flat_map(headers, fn {k, v} -> Enum.map(List.wrap(v), &{k, &1}) end)
     end
   end
 end

test/phoenix/sync/electric/client_adapter_test.exs

@@ -0,0 +1,46 @@
+defmodule Phoenix.Sync.Electric.ClientAdapterTest do
+  use ExUnit.Case, async: true
+
+  import Plug.Test
+
+  alias Phoenix.Sync.Electric.ClientAdapter
+
+  defmodule MockFetch do
+    def validate_opts(opts), do: {:ok, opts}
+
+    def fetch(request, parent: parent) do
+      send(parent, {:fetch_request, request})
+
+      %Electric.Client.Fetch.Response{
+        status: 200,
+        headers: %{},
+        body: ["[]"]
+      }
+    end
+  end
+
+  test "forwards request headers to sync server" do
+    {:ok, client} =
+      Electric.Client.new(
+        base_url: "elixir://#{inspect(__MODULE__.Fetch)}",
+        fetch: {MockFetch, parent: self()}
+      )
+
+    adapter = %ClientAdapter{client: client}
+
+    conn =
+      conn(:get, "/v1/shapes", %{})
+      |> Plug.Conn.put_req_header("my-header-1", "my-header-1-value-1")
+      |> Plug.Conn.prepend_req_headers([{"my-header-1", "my-header-1-value-2"}])
+      |> Plug.Conn.put_req_header("my-header-2", "my-header-2-value")
+
+    assert %{status: 200} = Phoenix.Sync.Adapter.PlugApi.call(adapter, conn, %{offset: -1})
+    assert_receive {:fetch_request, request}
+
+    assert request.headers == [
+             {"my-header-1", "my-header-1-value-1"},
+             {"my-header-1", "my-header-1-value-2"},
+             {"my-header-2", "my-header-2-value"}
+           ]
+  end
+end