feat: v2 serializer

edgurgel · edgurgel · commit ae956e0983d2 · 2025-11-05T10:42:14.000+13:00
diff --git a/.github/workflows/integration_tests.yml b/.github/workflows/integration_tests.yml
@@ -0,0 +1,33 @@
+name: Integration Tests
+on:
+  pull_request:
+    paths:
+      - "lib/**"
+      - "test/**"
+      - "config/**"
+      - "priv/**"
+      - "assets/**"
+      - "rel/**"
+      - "mix.exs"
+      - "Dockerfile"
+      - "run.sh"
+      - "docker-compose.test.yml"
+
+  push:
+    branches:
+      - main
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  tests:
+    name: Tests
+    runs-on: blacksmith-8vcpu-ubuntu-2404
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Run integration test
+        run: docker compose -f docker-compose.tests.yml up --abort-on-container-exit --exit-code-from test-runner
+
diff --git a/docker-compose.tests.yml b/docker-compose.tests.yml
@@ -0,0 +1,92 @@
+services:
+  # Supabase Realtime service
+  test_db:
+    image: supabase/postgres:14.1.0.105
+    container_name: test-realtime-db
+    ports:
+      - "5432:5432"
+    volumes:
+      - ./dev/postgres:/docker-entrypoint-initdb.d/
+    command: postgres -c config_file=/etc/postgresql/postgresql.conf
+    environment:
+      POSTGRES_HOST: /var/run/postgresql
+      POSTGRES_PASSWORD: postgres
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+  test_tenant_db:
+    image: supabase/postgres:14.1.0.105
+    container_name: test-tenant-db
+    ports:
+      - "5433:5432"
+    command: postgres -c config_file=/etc/postgresql/postgresql.conf
+    environment:
+      POSTGRES_HOST: /var/run/postgresql
+      POSTGRES_PASSWORD: postgres
+  test_realtime:
+    depends_on:
+      - test_db
+    build: .
+    container_name: test-realtime-server
+    ports:
+      - "4000:4000"
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+    environment:
+      PORT: 4000
+      DB_HOST: host.docker.internal
+      DB_PORT: 5432
+      DB_USER: postgres
+      DB_PASSWORD: postgres
+      DB_NAME: postgres
+      DB_ENC_KEY: 1234567890123456
+      DB_AFTER_CONNECT_QUERY: 'SET search_path TO _realtime'
+      API_JWT_SECRET: super-secret-jwt-token-with-at-least-32-characters-long
+      SECRET_KEY_BASE: UpNVntn3cDxHJpq99YMc1T1AQgQpc8kfYTuRgBiYa15BLrx8etQoXz3gZv1/u2oq
+      ERL_AFLAGS: -proto_dist inet_tcp
+      RLIMIT_NOFILE: 1000000
+      DNS_NODES: "''"
+      APP_NAME: realtime
+      RUN_JANITOR: true
+      JANITOR_INTERVAL: 60000
+      LOG_LEVEL: "info"
+      SEED_SELF_HOST: true
+    networks:
+      test-network:
+        aliases:
+          - realtime-dev.local
+          - realtime-dev.localhost
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:4000/"]
+      interval: 10s
+      timeout: 5s
+      retries: 5
+      start_period: 30s
+
+  # Deno test runner
+  test-runner:
+    image: denoland/deno:alpine-2.5.6
+    container_name: deno-test-runner
+    depends_on:
+      test_realtime:
+        condition: service_healthy
+      test_db:
+        condition: service_healthy
+    volumes:
+      - ./test/integration/tests.ts:/app/tests.ts:ro
+    working_dir: /app
+    command: >
+      sh -c "
+      echo 'Running tests...' &&
+      deno test tests.ts --allow-import --no-check --allow-read --allow-net --trace-leaks --allow-env=WS_NO_BUFFER_UTIL
+      "
+    networks:
+      - test-network
+    extra_hosts:
+      - "realtime-dev.localhost:host-gateway"
+
+networks:
+  test-network:
+    driver: bridge
diff --git a/lib/realtime_web/channels/realtime_channel/broadcast_handler.ex b/lib/realtime_web/channels/realtime_channel/broadcast_handler.ex
@@ -101,7 +101,7 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandler do
   end
 
   defp send_message(tenant_id, self_broadcast, tenant_topic, payload) do
-    broadcast = %Phoenix.Socket.Broadcast{topic: tenant_topic, event: @event_type, payload: payload}
+    broadcast = build_broadcast(tenant_topic, payload)
 
     if self_broadcast do
       TenantBroadcaster.pubsub_broadcast(
@@ -123,6 +123,20 @@ defmodule RealtimeWeb.RealtimeChannel.BroadcastHandler do
     end
   end
 
+  # Message payload was built by V2 Serializer which was originally UserBroadcast
+  defp build_broadcast(topic, {user_event, user_payload_encoding, user_payload}) do
+    %RealtimeWeb.Socket.UserBroadcast{
+      topic: topic,
+      user_event: user_event,
+      user_payload_encoding: user_payload_encoding,
+      user_payload: user_payload
+    }
+  end
+
+  defp build_broadcast(topic, payload) do
+    %Phoenix.Socket.Broadcast{topic: topic, event: @event_type, payload: payload}
+  end
+
   defp increment_rate_counter(%{assigns: %{policies: %Policies{broadcast: %BroadcastPolicies{write: false}}}} = socket) do
     socket
   end
diff --git a/lib/realtime_web/channels/realtime_channel/message_dispatcher.ex b/lib/realtime_web/channels/realtime_channel/message_dispatcher.ex
@@ -4,60 +4,74 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
   """
 
   require Logger
+  alias Phoenix.Socket.Broadcast
+  alias RealtimeWeb.Socket.UserBroadcast
 
   def fastlane_metadata(fastlane_pid, serializer, topic, log_level, tenant_id, replayed_message_ids \\ MapSet.new()) do
     {:rc_fastlane, fastlane_pid, serializer, topic, log_level, tenant_id, replayed_message_ids}
   end
 
+  @presence_diff "presence_diff"
+
   @doc """
   This dispatch function caches encoded messages if fastlane is used
   It also sends  an :update_rate_counter to the subscriber and it can conditionally log
-  """
-  @spec dispatch(list, pid, Phoenix.Socket.Broadcast.t()) :: :ok
-  def dispatch(subscribers, from, %Phoenix.Socket.Broadcast{event: event} = msg) do
-    # fastlane_pid is the actual socket transport pid
-    # This reduce caches the serialization and bypasses the channel process going straight to the
-    # transport process
-
-    message_id = message_id(msg.payload)
 
+  fastlane_pid is the actual socket transport pid
+  """
+  @spec dispatch(list, pid, Broadcast.t() | UserBroadcast.t()) :: :ok
+  def dispatch(subscribers, from, %Broadcast{event: @presence_diff} = msg) do
     {_cache, count} =
       Enum.reduce(subscribers, {%{}, 0}, fn
         {pid, _}, {cache, count} when pid == from ->
           {cache, count}
 
-        {pid, {:rc_fastlane, fastlane_pid, serializer, join_topic, log_level, tenant_id, replayed_message_ids}},
+        {_pid, {:rc_fastlane, fastlane_pid, serializer, join_topic, log_level, tenant_id, _replayed_message_ids}},
         {cache, count} ->
-          if already_replayed?(message_id, replayed_message_ids) do
-            # skip already replayed message
-            {cache, count}
-          else
-            if event != "presence_diff", do: send(pid, :update_rate_counter)
+          maybe_log(log_level, join_topic, msg, tenant_id)
 
-            maybe_log(log_level, join_topic, msg, tenant_id)
-
-            cache = do_dispatch(msg, fastlane_pid, serializer, join_topic, cache)
-            {cache, count + 1}
-          end
+          cache = do_dispatch(msg, fastlane_pid, serializer, join_topic, cache)
+          {cache, count + 1}
 
         {pid, _}, {cache, count} ->
           send(pid, msg)
           {cache, count}
       end)
 
     tenant_id = tenant_id(subscribers)
-    increment_presence_counter(tenant_id, event, count)
+    increment_presence_counter(tenant_id, msg.event, count)
 
     :ok
   end
 
-  defp increment_presence_counter(tenant_id, "presence_diff", count) when is_binary(tenant_id) do
-    tenant_id
-    |> Realtime.Tenants.presence_events_per_second_key()
-    |> Realtime.GenCounter.add(count)
-  end
+  def dispatch(subscribers, from, msg) do
+    message_id = message_id(msg)
 
-  defp increment_presence_counter(_tenant_id, _event, _count), do: :ok
+    _ =
+      Enum.reduce(subscribers, %{}, fn
+        {pid, _}, cache when pid == from ->
+          cache
+
+        {pid, {:rc_fastlane, fastlane_pid, serializer, join_topic, log_level, tenant_id, replayed_message_ids}},
+        cache ->
+          if already_replayed?(message_id, replayed_message_ids) do
+            # skip already replayed message
+            cache
+          else
+            send(pid, :update_rate_counter)
+
+            maybe_log(log_level, join_topic, msg, tenant_id)
+
+            do_dispatch(msg, fastlane_pid, serializer, join_topic, cache)
+          end
+
+        {pid, _}, cache ->
+          send(pid, msg)
+          cache
+      end)
+
+    :ok
+  end
 
   defp maybe_log(:info, join_topic, msg, tenant_id) do
     log = "Received message on #{join_topic} with payload: #{inspect(msg, pretty: true)}"
@@ -66,30 +80,53 @@ defmodule RealtimeWeb.RealtimeChannel.MessageDispatcher do
 
   defp maybe_log(_level, _join_topic, _msg, _tenant_id), do: :ok
 
-  defp message_id(%{"meta" => %{"id" => id}}), do: id
-  defp message_id(_), do: nil
-
-  defp already_replayed?(nil, _replayed_message_ids), do: false
-  defp already_replayed?(message_id, replayed_message_ids), do: MapSet.member?(replayed_message_ids, message_id)
-
   defp do_dispatch(msg, fastlane_pid, serializer, join_topic, cache) do
     case cache do
-      %{^serializer => encoded_msg} ->
+      %{^serializer => {:ok, encoded_msg}} ->
         send(fastlane_pid, encoded_msg)
         cache
 
+      %{^serializer => {:error, reason}} ->
+        # FIXME: log once when we try to serialize!
+        raise reason
+        cache
+
       %{} ->
         # Use the original topic that was joined without the external_id
         msg = %{msg | topic: join_topic}
-        encoded_msg = serializer.fastlane!(msg)
-        send(fastlane_pid, encoded_msg)
-        Map.put(cache, serializer, encoded_msg)
+
+        result =
+          with {:ok, encoded_msg} <- fastlane!(serializer, msg) do
+            send(fastlane_pid, encoded_msg)
+            {:ok, encoded_msg}
+          end
+
+        Map.put(cache, serializer, result)
     end
   end
 
-  defp tenant_id([{_pid, {:rc_fastlane, _, _, _, _, tenant_id, _}} | _]) do
-    tenant_id
+  defp fastlane!(Phoenix.Socket.V1.JSONSerializer = serializer, %UserBroadcast{} = msg) do
+    with {:ok, msg} <- UserBroadcast.convert_to_json_broadcast(msg) do
+      {:ok, serializer.fastlane!(msg)}
+    end
   end
 
+  defp fastlane!(serializer, msg), do: {:ok, serializer.fastlane!(msg)}
+
+  defp tenant_id([{_pid, {:rc_fastlane, _, _, _, _, tenant_id, _}} | _]), do: tenant_id
   defp tenant_id(_), do: nil
+
+  defp increment_presence_counter(tenant_id, "presence_diff", count) when is_binary(tenant_id) do
+    tenant_id
+    |> Realtime.Tenants.presence_events_per_second_key()
+    |> Realtime.GenCounter.add(count)
+  end
+
+  defp increment_presence_counter(_tenant_id, _event, _count), do: :ok
+
+  defp message_id(%Broadcast{payload: %{"meta" => %{"id" => id}}}), do: id
+  defp message_id(_), do: nil
+
+  defp already_replayed?(nil, _replayed_message_ids), do: false
+  defp already_replayed?(message_id, replayed_message_ids), do: MapSet.member?(replayed_message_ids, message_id)
 end
diff --git a/lib/realtime_web/endpoint.ex b/lib/realtime_web/endpoint.ex
@@ -25,11 +25,11 @@ defmodule RealtimeWeb.Endpoint do
       # the expense of potentially higher memory being used.
       active_n: 100,
       # Skip validating UTF8 for faster frame processing.
-      # Currently all text frames as handled only with JSON which already requires UTF-8
+      # Currently all text frames are handled only with JSON which already requires UTF-8
       validate_utf8: false,
       serializer: [
         {Phoenix.Socket.V1.JSONSerializer, "~> 1.0.0"},
-        {Phoenix.Socket.V2.JSONSerializer, "~> 2.0.0"}
+        {RealtimeWeb.Socket.V2Serializer, "~> 2.0.0"}
       ]
     ],
     longpoll: [
diff --git a/lib/realtime_web/plugs/auth_tenant.ex b/lib/realtime_web/plugs/auth_tenant.ex
@@ -42,6 +42,9 @@ defmodule RealtimeWeb.AuthTenant do
         [] ->
           nil
 
+        [""] ->
+          nil
+
         [value | _] ->
           [bearer, token] = value |> String.split(" ")
           bearer = String.downcase(bearer)
diff --git a/lib/realtime_web/socket/user_broadcast.ex b/lib/realtime_web/socket/user_broadcast.ex
@@ -0,0 +1,39 @@
+defmodule RealtimeWeb.Socket.UserBroadcast do
+  @moduledoc """
+  Defines a message sent from pubsub to channels and vice-versa.
+
+  The message format requires the following keys:
+
+    * `:topic` - The string topic or topic:subtopic pair namespace, for example "messages", "messages:123"
+    * `:user_event`- The string user event name, for example "my-event"
+    * `:user_payload_encoding`- :json or :binary
+    * `:user_payload` - The actual message payload
+
+  Optionally metadata which is a map to be JSON encoded
+  """
+
+  alias Phoenix.Socket.Broadcast
+
+  @type t :: %__MODULE__{}
+  defstruct topic: nil, user_event: nil, user_payload: nil, user_payload_encoding: nil, metadata: nil
+
+  @spec convert_to_json_broadcast(t) :: {:ok, %Broadcast{}} | {:error, String.t()}
+  def convert_to_json_broadcast(user_broadcast = %__MODULE__{user_payload_encoding: :json}) do
+    payload = %{
+      "event" => user_broadcast.user_event,
+      "payload" => Jason.Fragment.new(user_broadcast.user_payload),
+      "type" => "broadcast"
+    }
+
+    payload =
+      if user_broadcast.metadata do
+        Map.put(payload, "meta", user_broadcast.metadata)
+      else
+        payload
+      end
+
+    {:ok, %Broadcast{event: "broadcast", payload: payload, topic: user_broadcast.topic}}
+  end
+
+  def convert_to_json_broadcast(%__MODULE__{}), do: {:error, "User payload encoding is not JSON"}
+end
diff --git a/lib/realtime_web/socket/v2_serializer.ex b/lib/realtime_web/socket/v2_serializer.ex
diff --git a/test/integration/tests.ts b/test/integration/tests.ts
diff --git a/test/realtime_web/socket/v2_serializer_test.exs b/test/realtime_web/socket/v2_serializer_test.exs
