feat: setup supabase_realtime_admin

.github/workflows/integration_tests.yml

@@ -35,17 +35,25 @@ jobs:
     runs-on: blacksmith-8vcpu-ubuntu-2404
     env:
       POSTGRES_IMAGE: ${{ matrix.postgres_image }}
+      DB_USER: ${{ matrix.db_user }}
     strategy:
       fail-fast: false
       matrix:
-        postgres: [pg14, pg15, pg17]
+        postgres: [pg14, pg15, pg15_latest, pg17]
         include:
           - postgres: pg14
-            postgres_image: supabase/postgres:14.1.0.85
+            postgres_image: supabase/postgres:14.1.0.82
+            db_user: supabase_admin
+          # missing supautils.policy_grants
           - postgres: pg15
-            postgres_image: supabase/postgres:15.14.1.127
+            postgres_image: supabase/postgres:15.1.0.1
+            db_user: supabase_admin
+          - postgres: pg15_latest
+            postgres_image: supabase/postgres:15.14.1.129
+            db_user: supabase_realtime_admin
           - postgres: pg17
             postgres_image: supabase/postgres:17.6.1.127
+            db_user: supabase_realtime_admin
 
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2

.github/workflows/tests.yml

@@ -9,7 +9,9 @@ on:
       - "assets/**"
       - "rel/**"
       - "native/**"
+      - "dev/**"
       - "mix.exs"
+      - "compose.dbs.yml"
       - "mix.lock"
       - "Dockerfile"
       - "run.sh"
@@ -35,18 +37,26 @@ jobs:
     runs-on: blacksmith-8vcpu-ubuntu-2404
     env:
       POSTGRES_IMAGE: ${{ matrix.postgres_image }}
+      DB_USER: ${{ matrix.db_user }}
     strategy:
       fail-fast: false
       matrix:
         partition: [1, 2, 3, 4]
-        postgres: [pg14, pg15, pg17]
+        postgres: [pg14, pg15, pg15_latest, pg17]
         include:
           - postgres: pg14
-            postgres_image: supabase/postgres:14.1.0.85
+            postgres_image: supabase/postgres:14.1.0.82
+            db_user: supabase_admin
+          # missing supautils.policy_grants
           - postgres: pg15
-            postgres_image: supabase/postgres:15.14.1.127
+            postgres_image: supabase/postgres:15.1.0.1
+            db_user: supabase_admin
+          - postgres: pg15_latest
+            postgres_image: supabase/postgres:15.14.1.129
+            db_user: supabase_realtime_admin
           - postgres: pg17
             postgres_image: supabase/postgres:17.6.1.127
+            db_user: supabase_realtime_admin
 
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2

README.md

@@ -67,6 +67,15 @@ Once your environment is up and running, check out the following docs to customi
 - [ERROR_CODES.md](ERROR_CODES.md) - list of operational codes
 - [OBSERVABILITY_METRICS.md](OBSERVABILITY_METRICS.md) - monitoring information
 
+## Postgres compatibility
+
+| `supabase/postgres` version       | Role                      | Status                                                                                                                                                                                                                                      |
+| --------------------------------- | ------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| < 14                              | -                         | Not officially supported.                                                                                                                                                                                                                   |
+| 14.x                              | `supabase_admin`          | Requires superuser: `log_min_messages` can only be set by a superuser; supautils doesn't expose per-parameter delegation on this version. On <= 14.5, `realtime.broadcast_changes(...)` called from a trigger via `PERFORM` is unsupported. |
+| 15.x < 15.14.1.018                | `supabase_admin`          | Requires superuser: `supautils.policy_grants` on `realtime.subscription` is missing until [supabase/postgres@1b916920](https://github.com/supabase/postgres/commit/1b916920).                                                               |
+| 15.x >= 15.14.1.018, 16.x, 17.x   | `supabase_realtime_admin` | No superuser needed. Role must have `REPLICATION` and policies are managed by supautils.                                                                                                                                                    |
+
 ## Contributing
 
 See [CONTRIBUTING.md](CONTRIBUTING.md)

compose.dbs.yml

@@ -4,7 +4,8 @@ services:
     ports:
       - "5432:5432"
     volumes:
-      - ./dev/postgres/zz-supabase-schema.sql:/docker-entrypoint-initdb.d/zz-supabase-schema.sql
+      - ./dev/postgres/za-permit-supabase-admin.sh:/docker-entrypoint-initdb.d/za-permit-supabase-admin.sh
+      - ./dev/postgres/zb-supabase-schema.sql:/docker-entrypoint-initdb.d/zb-supabase-schema.sql
     command: postgres -c config_file=/etc/postgresql/postgresql.conf
     environment:
       POSTGRES_HOST: /var/run/postgresql
@@ -20,7 +21,8 @@ services:
     ports:
       - "5433:5432"
     volumes:
-      - ./dev/postgres/zz-supabase-schema.sql:/docker-entrypoint-initdb.d/zz-supabase-schema.sql
+      - ./dev/postgres/za-permit-supabase-admin.sh:/docker-entrypoint-initdb.d/za-permit-supabase-admin.sh
+      - ./dev/postgres/zb-supabase-schema.sql:/docker-entrypoint-initdb.d/zb-supabase-schema.sql
     command: postgres -c config_file=/etc/postgresql/postgresql.conf
     environment:
       POSTGRES_HOST: /var/run/postgresql

compose.tests.yml

@@ -6,7 +6,8 @@ services:
     ports:
       - "5532:5432"
     volumes:
-      - ./dev/postgres:/docker-entrypoint-initdb.d/
+      - ./dev/postgres/za-permit-supabase-admin.sh:/docker-entrypoint-initdb.d/za-permit-supabase-admin.sh
+      - ./dev/postgres/zb-supabase-schema.sql:/docker-entrypoint-initdb.d/zb-supabase-schema.sql
     command: postgres -c config_file=/etc/postgresql/postgresql.conf
     environment:
       POSTGRES_HOST: /var/run/postgresql
@@ -29,7 +30,7 @@ services:
       PORT: 4100
       DB_HOST: host.docker.internal
       DB_PORT: 5532
-      DB_USER: supabase_admin
+      DB_USER: ${DB_USER:-supabase_realtime_admin}
       DB_PASSWORD: postgres
       DB_NAME: postgres
       DB_ENC_KEY: 1234567890123456

config/runtime.exs

@@ -28,7 +28,7 @@ db_replica_host = System.get_env("DB_REPLICA_HOST")
 db_replica_pool_size = Env.get_integer("DB_REPLICA_POOL_SIZE", 5)
 db_ssl = Env.get_boolean("DB_SSL", false)
 db_ssl_ca_cert = System.get_env("DB_SSL_CA_CERT")
-db_user = System.get_env("DB_USER", "supabase_admin")
+db_user = System.get_env("DB_USER", "supabase_realtime_admin")
 disable_healthcheck_logging = Env.get_boolean("DISABLE_HEALTHCHECK_LOGGING", false)
 dns_nodes = System.get_env("DNS_NODES")
 gen_rpc_compress = Env.get_integer("GEN_RPC_COMPRESS", 0)

config/test.exs

@@ -15,7 +15,7 @@ for repo <- [
       Realtime.Repo.Replica.SanJose
     ] do
   config :realtime, repo,
-    username: "supabase_admin",
+    username: "supabase_realtime_admin",
     password: "postgres",
     database: "realtime_test#{partition}",
     hostname: "127.0.0.1",

dev/postgres/za-permit-supabase-admin.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+# Allow peer auth as supabase_admin so the next init .sql can `\connect -` to it without a password
+set -euo pipefail
+
+echo "supabase_map postgres supabase_admin" >> "${PGDATA}/pg_ident.conf"
+printf 'local all supabase_admin peer map=supabase_map\n%s' "$(cat "${PGDATA}/pg_hba.conf")" > "${PGDATA}/pg_hba.conf.new"
+mv "${PGDATA}/pg_hba.conf.new" "${PGDATA}/pg_hba.conf"
+pg_ctl reload -D "${PGDATA}" >/dev/null

dev/postgres/zb-supabase-schema.sql

@@ -0,0 +1,26 @@
+\connect - supabase_admin
+
+do $$
+begin
+  if not exists (select from pg_roles where rolname = 'supabase_realtime_admin') then
+    create user supabase_realtime_admin noinherit createrole login replication password 'postgres';
+  end if;
+end$$;
+
+create schema if not exists _realtime;
+
+alter user supabase_realtime_admin set search_path = public, extensions, realtime;
+grant create on database postgres to supabase_realtime_admin;
+do $$
+begin
+  if current_setting('server_version_num')::int >= 150000 then
+    execute 'grant set on parameter log_min_messages to supabase_realtime_admin';
+  end if;
+end$$;
+grant anon, authenticated, service_role to supabase_realtime_admin;
+grant create, usage on schema public to supabase_realtime_admin;
+grant usage on schema extensions to supabase_realtime_admin;
+grant usage on schema auth to supabase_realtime_admin;
+grant execute on all functions in schema auth to supabase_realtime_admin;
+grant all on schema realtime to supabase_realtime_admin with grant option;
+grant create, usage on schema _realtime to supabase_realtime_admin;

lib/realtime/tenants.ex

@@ -476,7 +476,14 @@ defmodule Realtime.Tenants do
   Checks if migrations for a given tenant need to run.
   """
   @spec run_migrations?(Tenant.t() | integer()) :: boolean()
-  def run_migrations?(%Tenant{} = tenant), do: run_migrations?(tenant.migrations_ran)
+  def run_migrations?(%Tenant{} = tenant) do
+    available_migrations =
+      tenant.external_id
+      |> Migrations.migrations()
+      |> Enum.count()
+
+    tenant.migrations_ran < available_migrations
+  end
 
   def run_migrations?(migrations_ran) when is_integer(migrations_ran),
     do: migrations_ran < Enum.count(Migrations.migrations())

lib/realtime/tenants/migrations.ex

@@ -7,6 +7,7 @@ defmodule Realtime.Tenants.Migrations do
 
   alias Realtime.Tenants
   alias Realtime.Database
+  alias Realtime.FeatureFlags
   alias Realtime.Registry.Unique
   alias Realtime.Repo
   alias Realtime.Api.Tenant
@@ -85,7 +86,9 @@ defmodule Realtime.Tenants.Migrations do
     FilterActionPostgresChanges,
     FixByteaDoubleEncodingInCast,
     ListChangesWithSlotCount,
-    AddBinaryPayloadToMessages
+    SubscriptionCheckFiltersUsePgAttribute,
+    AddBinaryPayloadToMessages,
+    SetupSupabaseRealtimeAdmin
   }
 
   @migrations [
@@ -158,7 +161,9 @@ defmodule Realtime.Tenants.Migrations do
     {20_251_120_215_549, FilterActionPostgresChanges},
     {20_260_218_120_000, FixByteaDoubleEncodingInCast},
     {20_260_326_120_000, ListChangesWithSlotCount},
-    {20_260_514_120_000, AddBinaryPayloadToMessages}
+    {20_260_506_120_000, SubscriptionCheckFiltersUsePgAttribute},
+    {20_260_514_120_000, AddBinaryPayloadToMessages},
+    {20_260_515_120_000, SetupSupabaseRealtimeAdmin}
   ]
 
   defstruct [:tenant_external_id, :settings, migrations_ran: 0]
@@ -218,7 +223,7 @@ defmodule Realtime.Tenants.Migrations do
       :ok ->
         Task.Supervisor.async_nolink(__MODULE__.TaskSupervisor, Api, :update_migrations_ran, [
           tenant_external_id,
-          Enum.count(@migrations)
+          Enum.count(migrations(tenant_external_id))
         ])
 
         :ignore
@@ -251,7 +256,7 @@ defmodule Realtime.Tenants.Migrations do
 
         try do
           opts = [all: true, prefix: "realtime", dynamic_repo: repo]
-          result = Ecto.Migrator.run(Repo, @migrations, :up, opts)
+          result = Ecto.Migrator.run(Repo, migrations(tenant_external_id), :up, opts)
           Telemetry.stop(event, start_time, Map.put(metadata, :migrations_executed, length(result)))
         rescue
           error ->
@@ -294,14 +299,18 @@ defmodule Realtime.Tenants.Migrations do
       end_timestamp = Date.to_string(Date.add(date, 1))
 
       Database.transaction(db_conn_pid, fn conn ->
-        query = """
+        create = """
         CREATE TABLE IF NOT EXISTS realtime.#{partition_name}
         PARTITION OF realtime.messages
-        FOR VALUES FROM ('#{start_timestamp}') TO ('#{end_timestamp}');
+        FOR VALUES FROM ('#{start_timestamp}') TO ('#{end_timestamp}')
         """
 
-        case Postgrex.query(conn, query, []) do
-          {:ok, _} -> Logger.debug("Partition #{partition_name} created")
+        alter_owner = "ALTER TABLE realtime.#{partition_name} OWNER TO supabase_realtime_admin"
+
+        with {:ok, _} <- Postgrex.query(conn, create, []),
+             {:ok, _} <- Postgrex.query(conn, alter_owner, []) do
+          Logger.debug("Partition #{partition_name} created")
+        else
           {:error, %Postgrex.Error{postgres: %{code: :duplicate_table}}} -> :ok
           {:error, error} -> log_error("PartitionCreationFailed", error)
         end
@@ -311,5 +320,21 @@ defmodule Realtime.Tenants.Migrations do
     :ok
   end
 
-  def migrations(), do: @migrations
+  @doc """
+  Returns the migrations to run.
+  """
+  @spec migrations(String.t() | nil) :: [{pos_integer(), module()}]
+  def migrations(tenant_external_id \\ nil) do
+    Enum.filter(@migrations, fn {_version, module} -> migration_enabled?(module, tenant_external_id) end)
+  end
+
+  defp migration_enabled?(SetupSupabaseRealtimeAdmin, nil = _tenant_external_id) do
+    FeatureFlags.enabled?("use_supabase_realtime_admin")
+  end
+
+  defp migration_enabled?(SetupSupabaseRealtimeAdmin, tenant_external_id) when is_binary(tenant_external_id) do
+    FeatureFlags.enabled?("use_supabase_realtime_admin", tenant_external_id)
+  end
+
+  defp migration_enabled?(_migration, _tenant_external_id), do: true
 end

lib/realtime/tenants/repo/migrations/20240401105812_create_realtime_admin_and_move_ownership.ex

@@ -30,6 +30,13 @@ defmodule Realtime.Tenants.Migrations.CreateRealtimeAdminAndMoveOwnership do
     execute("ALTER table realtime.presences OWNER TO supabase_realtime_admin")
     execute("ALTER function realtime.channel_name() owner to supabase_realtime_admin")
 
-    execute("GRANT supabase_realtime_admin TO postgres")
+    execute("""
+    DO $$
+    BEGIN
+      IF (SELECT rolsuper FROM pg_roles WHERE rolname = current_user) THEN
+        GRANT supabase_realtime_admin TO postgres;
+      END IF;
+    END $$;
+    """)
   end
 end