feat: broadcast storage

lib/realtime/api/message.ex

@@ -12,10 +12,11 @@ defmodule Realtime.Api.Message do
   @timestamps_opts [type: :naive_datetime_usec]
   schema "messages" do
     field(:topic, :string)
-    field(:extension, Ecto.Enum, values: [:broadcast, :presence])
+    field(:extension, Ecto.Enum, values: [:broadcast, :presence, :persistence])
     field(:payload, :map)
     field(:event, :string)
     field(:private, :boolean)
+    field(:broadcasted_at, :naive_datetime_usec)
 
     timestamps()
   end
@@ -28,6 +29,7 @@ defmodule Realtime.Api.Message do
       :payload,
       :event,
       :private,
+      :broadcasted_at,
       :inserted_at,
       :updated_at
     ])

lib/realtime/messages.ex

@@ -4,12 +4,49 @@ defmodule Realtime.Messages do
   """
 
   alias Realtime.Api.Message
+  alias Realtime.Tenants.Repo
 
   import Ecto.Query, only: [from: 2]
 
   @hard_limit 25
   @default_timeout 5_000
 
+  @doc """
+  Persists a broadcast sent over WebSocket for `topic` as a `persistence` message.
+
+  Only called after the sender's `persistence` policy authorized it, so the persisted row is always
+  private. Bypasses RLS because that authorization already happened on the broadcast, and sets
+  `broadcasted_at` to avoid re-broadcasting the message twice.
+
+  Automatically uses RPC if the database connection is not on the same node.
+  """
+  @spec store(DBConnection.conn(), String.t(), String.t(), term()) ::
+          {:ok, binary()} | {:error, any()} | {:error, :rpc_error, term}
+  def store(conn, topic, event, payload) when node(conn) == node() do
+    insert(conn, topic, event, payload)
+  end
+
+  def store(conn, topic, event, payload) do
+    Realtime.GenRpc.call(node(conn), __MODULE__, :store, [conn, topic, event, payload], key: topic)
+  end
+
+  defp insert(conn, topic, event, payload) do
+    changeset =
+      Message.changeset(%Message{}, %{
+        topic: topic,
+        extension: :persistence,
+        event: event,
+        payload: payload,
+        private: true,
+        broadcasted_at: NaiveDateTime.utc_now(:microsecond)
+      })
+
+    case Repo.insert(conn, changeset, Message) do
+      {:ok, %Message{id: id}} -> {:ok, id}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
   @doc """
   Fetch last `limit ` messages for a given `topic` inserted after `since`
 
@@ -53,7 +90,7 @@ defmodule Realtime.Messages do
         where:
           m.topic == ^topic and
             m.private == true and
-            m.extension == :broadcast and
+            m.extension in [:broadcast, :persistence] and
             m.inserted_at >= ^since and
             m.inserted_at < ^now,
         limit: ^limit,

lib/realtime/tenants/authorization.ex

@@ -320,26 +320,35 @@ defmodule Realtime.Tenants.Authorization do
   end
 
   defp check_write_policies(conn, authorization_context, extensions, policies) do
+    with {:ok, policies} <- check_extension_write_policies(conn, authorization_context, extensions, policies),
+         {:ok, persistence_write?} <- authorize_write(conn, authorization_context, :persistence) do
+      {:ok, Policies.update_policies(policies, :persistence, :write, persistence_write?)}
+    end
+  end
+
+  defp check_extension_write_policies(conn, authorization_context, extensions, policies) do
     Enum.reduce_while(@all_extensions, {:ok, policies}, fn extension, {:ok, acc} ->
       if extension in extensions do
-        changeset = Message.changeset(%Message{}, %{topic: authorization_context.topic, extension: extension})
-
-        case Repo.insert(conn, changeset, Message, mode: :savepoint, returning: false) do
-          {:ok, _} ->
-            {:cont, {:ok, Policies.update_policies(acc, extension, :write, true)}}
-
-          {:error, %Postgrex.Error{postgres: %{code: :insufficient_privilege}}} ->
-            {:cont, {:ok, Policies.update_policies(acc, extension, :write, false)}}
-
-          {:error, reason} ->
-            {:halt, {:error, reason}}
+        case authorize_write(conn, authorization_context, extension) do
+          {:ok, allowed?} -> {:cont, {:ok, Policies.update_policies(acc, extension, :write, allowed?)}}
+          {:error, reason} -> {:halt, {:error, reason}}
         end
       else
         {:cont, {:ok, Policies.update_policies(acc, extension, :write, false)}}
       end
     end)
   end
 
+  defp authorize_write(conn, authorization_context, extension) do
+    changeset = Message.changeset(%Message{}, %{topic: authorization_context.topic, extension: extension})
+
+    case Repo.insert(conn, changeset, Message, mode: :savepoint, returning: false) do
+      {:ok, _} -> {:ok, true}
+      {:error, %Postgrex.Error{postgres: %{code: :insufficient_privilege}}} -> {:ok, false}
+      {:error, reason} -> {:error, reason}
+    end
+  end
+
   defp rate_counter(tenant_id) do
     %Tenant{} = tenant = Realtime.Tenants.Cache.get_tenant_by_external_id(tenant_id)
     rate_counter = Realtime.Tenants.authorization_errors_per_second_rate(tenant)

lib/realtime/tenants/authorization/policies.ex

@@ -2,20 +2,24 @@ defmodule Realtime.Tenants.Authorization.Policies do
   @moduledoc """
   Policies structure that holds the required authorization information for a given connection.
 
-  Currently there are two types of policies:
+  Currently there are three types of policies:
   * Realtime.Tenants.Authorization.Policies.BroadcastPolicies - Used to store the access to Broadcast feature on a given Topic
   * Realtime.Tenants.Authorization.Policies.PresencePolicies - Used to store the access to Presence feature on a given Topic
+  * Realtime.Tenants.Authorization.Policies.PersistencePolicies - Used to store whether messages may be persisted on a given Topic
   """
 
   alias Realtime.Tenants.Authorization.Policies.BroadcastPolicies
   alias Realtime.Tenants.Authorization.Policies.PresencePolicies
+  alias Realtime.Tenants.Authorization.Policies.PersistencePolicies
 
   defstruct broadcast: %BroadcastPolicies{},
-            presence: %PresencePolicies{}
+            presence: %PresencePolicies{},
+            persistence: %PersistencePolicies{}
 
   @type t :: %__MODULE__{
           broadcast: BroadcastPolicies.t(),
-          presence: PresencePolicies.t()
+          presence: PresencePolicies.t(),
+          persistence: PersistencePolicies.t()
         }
 
   @doc """

lib/realtime/tenants/authorization/policies/persistence_policies.ex

@@ -0,0 +1,10 @@
+defmodule Realtime.Tenants.Authorization.Policies.PersistencePolicies do
+  @moduledoc """
+  PersistencePolicies structure that holds whether a connection is authorized to persist messages for a given Topic.
+  """
+  defstruct write: nil
+
+  @type t :: %__MODULE__{
+          write: boolean() | nil
+        }
+end

lib/realtime/tenants/migrations.ex

@@ -98,7 +98,8 @@ defmodule Realtime.Tenants.Migrations do
     {20_260_626_120_000, Migrations.ReAddPostgrestFilterOps},
     {20_260_706_120_000, Migrations.GrantCheckEqualityOp5Arg},
     {20_260_707_120_000, Migrations.RestrictRealtimeSchema},
-    {20_260_709_120_000, Migrations.FixApplyRlsFilterRoleLeak}
+    {20_260_709_120_000, Migrations.FixApplyRlsFilterRoleLeak},
+    {20_260_714_120_000, Migrations.AddBroadcastStorage}
   ]
 
   defstruct [:tenant_external_id, :settings, migrations_ran: 0]

lib/realtime/tenants/replication_connection.ex

@@ -1,20 +1,6 @@
 defmodule Realtime.Tenants.ReplicationConnection do
   @moduledoc """
   ReplicationConnection it's the module that provides a way to stream data from a PostgreSQL database using logical replication.
-
-  ## Struct parameters
-  * `connection_opts` - The connection options to connect to the database.
-  * `table` - The table to replicate. If `:all` is passed, it will replicate all tables.
-  * `schema` - The schema of the table to replicate. If not provided, it will use the `public` schema. If `:all` is passed, this option is ignored.
-  * `opts` - The options to pass to this module
-  * `step` - The current step of the replication process
-  * `publication_name` - The name of the publication to create. If not provided, it will use the schema and table name.
-  * `replication_slot_name` - The name of the replication slot to create. If not provided, it will use the schema and table name.
-  * `output_plugin` - The output plugin to use. Default is `pgoutput`.
-  * `proto_version` - The protocol version to use. Default is `2`.
-  * `handler_module` - The module that will handle the data received from the replication stream.
-  * `metadata` - The metadata to pass to the handler module.
-
   """
   use Postgrex.ReplicationConnection
   use Realtime.Logs
@@ -39,6 +25,20 @@ defmodule Realtime.Tenants.ReplicationConnection do
 
   @default_query_timeout :timer.minutes(4)
 
+  @typedoc """
+  * `tenant_id` - The tenant this connection replicates for.
+  * `opts` - Reserved, defaults to `[]`.
+  * `step` - The current step of the replication process.
+  * `publication_name` - The name of the publication to create. Defaults to `"supabase_\#{schema}_\#{table}_publication"` if not provided.
+  * `replication_slot_name` - The name of the replication slot to create. Defaults to `"supabase_\#{schema}_\#{table}_replication_slot_\#{slot_suffix()}"` if not provided.
+  * `output_plugin` - The output plugin to use. Default is `pgoutput`.
+  * `proto_version` - The protocol version to use. Default is `2`.
+  * `relations` - Cache of decoded relation (table) metadata, keyed by relation id, populated as `Relation` messages stream in.
+  * `buffer` - Reserved, defaults to `[]`.
+  * `monitored_pid` - The process this connection is linked to; the connection stops when it dies.
+  * `latency_committed_at` - Timestamp of the last commit, used to compute broadcast latency.
+  * `query_timeout` - Timeout for queries run during the replication setup steps.
+  """
   @type t :: %__MODULE__{
           tenant_id: String.t(),
           opts: Keyword.t(),
@@ -409,6 +409,7 @@ defmodule Realtime.Tenants.ReplicationConnection do
 
     with %{columns: columns} <- Map.get(relations, relation_id),
          to_broadcast = tuple_to_map(tuple_data, columns),
+         :ok <- check_not_already_broadcasted(to_broadcast),
          {:ok, inserted_at} <- get_or_error(to_broadcast, "inserted_at", :inserted_at_missing),
          {:ok, event} <- get_or_error(to_broadcast, "event", :event_missing),
          {:ok, id} <- get_or_error(to_broadcast, "id", :id_missing),
@@ -449,6 +450,9 @@ defmodule Realtime.Tenants.ReplicationConnection do
 
       {:noreply, state}
     else
+      {:error, :already_broadcasted} ->
+        {:noreply, state}
+
       {:error, error} ->
         log_error("UnableToBroadcastChanges", error)
         {:noreply, state}
@@ -505,6 +509,10 @@ defmodule Realtime.Tenants.ReplicationConnection do
     end
   end
 
+  defp check_not_already_broadcasted(%{"broadcasted_at" => nil}), do: :ok
+  defp check_not_already_broadcasted(%{"broadcasted_at" => _}), do: {:error, :already_broadcasted}
+  defp check_not_already_broadcasted(_), do: :ok
+
   defp get_or_error(map, key, error_type) do
     case Map.get(map, key) do
       nil -> {:error, error_type}

lib/realtime/tenants/repo.ex

@@ -106,14 +106,6 @@ defmodule Realtime.Tenants.Repo do
     end
   end
 
-  defp result_to_single_struct(
-         {:error, %Postgrex.Error{postgres: %{code: :unique_violation, constraint: "channels_name_index"}}},
-         _struct,
-         changeset
-       ) do
-    Ecto.Changeset.add_error(changeset, :name, "has already been taken")
-  end
-
   defp result_to_single_struct({:error, _} = error, _, _), do: error
 
   defp result_to_single_struct({:ok, %Postgrex.Result{rows: []}}, _, _) do

lib/realtime/tenants/repo/migrations/20260714120000_add_broadcast_storage.ex

@@ -0,0 +1,12 @@
+defmodule Realtime.Tenants.Migrations.AddBroadcastStorage do
+  @moduledoc false
+  use Ecto.Migration
+
+  def up do
+    execute("ALTER TABLE realtime.messages ADD COLUMN IF NOT EXISTS broadcasted_at timestamp")
+  end
+
+  def down do
+    execute("ALTER TABLE realtime.messages DROP COLUMN IF EXISTS broadcasted_at")
+  end
+end

lib/realtime_web/channels/realtime_channel.ex

@@ -463,7 +463,7 @@ defmodule RealtimeWeb.RealtimeChannel do
   end
 
   def handle_in("broadcast", payload, %{assigns: %{private?: false}} = socket) do
-    BroadcastHandler.handle(payload, socket)
+    BroadcastHandler.handle(payload, nil, socket)
   end
 
   def handle_in("presence", payload, %{assigns: %{private?: true}} = socket) do