feat: setup supabase_realtime_admin

.github/workflows/tests.yml

@@ -9,7 +9,9 @@ on:
       - "assets/**"
       - "rel/**"
       - "native/**"
+      - "dev/**"
       - "mix.exs"
+      - "compose.dbs.yml"
       - "mix.lock"
       - "Dockerfile"
       - "run.sh"

compose.dbs.yml

@@ -4,7 +4,7 @@ services:
     ports:
       - "5432:5432"
     volumes:
-      - ./dev/postgres/00-supabase-schema.sql:/docker-entrypoint-initdb.d/00-supabase-schema.sql
+      - ./dev/postgres/01-realtime-setup.sql:/docker-entrypoint-initdb.d/init-scripts/01-realtime-setup.sql
     command: postgres -c config_file=/etc/postgresql/postgresql.conf
     environment:
       POSTGRES_HOST: /var/run/postgresql
@@ -19,6 +19,8 @@ services:
     image: ${POSTGRES_IMAGE:-supabase/postgres:17.6.1.074}
     ports:
       - "5433:5432"
+    volumes:
+      - ./dev/postgres/01-realtime-setup.sql:/docker-entrypoint-initdb.d/init-scripts/01-realtime-setup.sql
     command: postgres -c config_file=/etc/postgresql/postgresql.conf
     environment:
       POSTGRES_HOST: /var/run/postgresql

compose.tests.yml

@@ -6,7 +6,7 @@ services:
     ports:
       - "5532:5432"
     volumes:
-      - ./dev/postgres:/docker-entrypoint-initdb.d/
+      - ./dev/postgres/01-realtime-setup.sql:/docker-entrypoint-initdb.d/init-scripts/01-realtime-setup.sql
     command: postgres -c config_file=/etc/postgresql/postgresql.conf
     environment:
       POSTGRES_HOST: /var/run/postgresql
@@ -29,7 +29,7 @@ services:
       PORT: 4100
       DB_HOST: host.docker.internal
       DB_PORT: 5532
-      DB_USER: supabase_admin
+      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/01-realtime-setup.sql

@@ -0,0 +1,24 @@
+-- dev/test
+
+CREATE SCHEMA IF NOT EXISTS realtime AUTHORIZATION supabase_admin;
+CREATE SCHEMA IF NOT EXISTS _realtime;
+
+DO $$
+BEGIN
+  IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'supabase_realtime_admin') THEN
+    CREATE USER supabase_realtime_admin NOINHERIT CREATEROLE LOGIN REPLICATION;
+  END IF;
+END $$;
+
+ALTER USER supabase_realtime_admin WITH LOGIN REPLICATION PASSWORD 'postgres';
+ALTER USER supabase_realtime_admin SET search_path = public, extensions, realtime;
+GRANT CREATE ON DATABASE postgres TO supabase_realtime_admin;
+GRANT SET ON PARAMETER log_min_messages TO supabase_realtime_admin;
+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 USAGE ON SCHEMA realtime TO postgres, anon, authenticated, service_role;
+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
@@ -84,7 +85,9 @@ defmodule Realtime.Tenants.Migrations do
     AddActionToSubscriptions,
     FilterActionPostgresChanges,
     FixByteaDoubleEncodingInCast,
-    ListChangesWithSlotCount
+    ListChangesWithSlotCount,
+    SubscriptionCheckFiltersUsePgAttribute,
+    SetupSupabaseRealtimeAdmin
   }
 
   @migrations [
@@ -156,7 +159,9 @@ defmodule Realtime.Tenants.Migrations do
     {20_251_120_212_548, AddActionToSubscriptions},
     {20_251_120_215_549, FilterActionPostgresChanges},
     {20_260_218_120_000, FixByteaDoubleEncodingInCast},
-    {20_260_326_120_000, ListChangesWithSlotCount}
+    {20_260_326_120_000, ListChangesWithSlotCount},
+    {20_260_506_120_000, SubscriptionCheckFiltersUsePgAttribute},
+    {20_260_511_170_200, SetupSupabaseRealtimeAdmin}
   ]
 
   defstruct [:tenant_external_id, :settings, migrations_ran: 0]
@@ -216,7 +221,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
@@ -249,7 +254,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 ->
@@ -309,5 +314,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

lib/realtime/tenants/repo/migrations/20260506120000_subscription_check_filters_use_pg_attribute.ex

@@ -0,0 +1,69 @@
+defmodule Realtime.Tenants.Migrations.SubscriptionCheckFiltersUsePgAttribute do
+  @moduledoc false
+
+  use Ecto.Migration
+
+  def change do
+    execute("""
+    create or replace function realtime.subscription_check_filters()
+        returns trigger
+        language plpgsql
+    as $$
+    declare
+        col_names text[] = coalesce(
+                array_agg(a.attname order by a.attnum),
+                '{}'::text[]
+            )
+            from
+                pg_catalog.pg_attribute a
+            where
+                a.attrelid = new.entity
+                and a.attnum > 0
+                and not a.attisdropped
+                and pg_catalog.has_column_privilege(
+                    (new.claims ->> 'role'),
+                    a.attrelid,
+                    a.attnum,
+                    'SELECT'
+                );
+        filter realtime.user_defined_filter;
+        col_type regtype;
+
+        in_val jsonb;
+    begin
+        for filter in select * from unnest(new.filters) loop
+            if not filter.column_name = any(col_names) then
+                raise exception 'invalid column for filter %', filter.column_name;
+            end if;
+
+            col_type = (
+                select atttypid::regtype
+                from pg_catalog.pg_attribute
+                where attrelid = new.entity
+                      and attname = filter.column_name
+            );
+            if col_type is null then
+                raise exception 'failed to lookup type for column %', filter.column_name;
+            end if;
+
+            if filter.op = 'in'::realtime.equality_op then
+                in_val = realtime.cast(filter.value, (col_type::text || '[]')::regtype);
+                if coalesce(jsonb_array_length(in_val), 0) > 100 then
+                    raise exception 'too many values for `in` filter. Maximum 100';
+                end if;
+            else
+                perform realtime.cast(filter.value, col_type);
+            end if;
+        end loop;
+
+        new.filters = coalesce(
+            array_agg(f order by f.column_name, f.op, f.value),
+            '{}'
+        ) from unnest(new.filters) f;
+
+        return new;
+    end;
+    $$;
+    """)
+  end
+end

lib/realtime/tenants/repo/migrations/20260511170200_setup_supabase_realtime_admin.ex

@@ -0,0 +1,73 @@
+defmodule Realtime.Tenants.Migrations.SetupSupabaseRealtimeAdmin do
+  @moduledoc false
+
+  use Ecto.Migration
+
+  def change do
+    execute("""
+    DO $$
+    BEGIN
+      IF (SELECT rolsuper FROM pg_roles WHERE rolname = current_user) THEN
+        ALTER ROLE supabase_realtime_admin WITH NOINHERIT CREATEROLE LOGIN REPLICATION;
+        ALTER ROLE supabase_realtime_admin SET search_path = public, extensions, realtime;
+        GRANT CREATE ON DATABASE postgres TO supabase_realtime_admin;
+        GRANT SET ON PARAMETER log_min_messages TO supabase_realtime_admin;
+        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 USAGE ON SCHEMA realtime TO postgres, anon, authenticated, service_role;
+        GRANT ALL ON SCHEMA realtime TO supabase_realtime_admin WITH GRANT OPTION;
+      END IF;
+    END $$;
+    """)
+
+    execute("ALTER TABLE realtime.messages OWNER TO supabase_realtime_admin")
+    execute("ALTER TABLE realtime.subscription OWNER TO supabase_realtime_admin")
+    execute("ALTER TYPE realtime.action OWNER TO supabase_realtime_admin")
+    execute("ALTER TYPE realtime.equality_op OWNER TO supabase_realtime_admin")
+    execute("ALTER TYPE realtime.user_defined_filter OWNER TO supabase_realtime_admin")
+    execute("ALTER TYPE realtime.wal_column OWNER TO supabase_realtime_admin")
+    execute("ALTER TYPE realtime.wal_rls OWNER TO supabase_realtime_admin")
+    execute("ALTER FUNCTION realtime.apply_rls(jsonb, integer) OWNER TO supabase_realtime_admin")
+
+    execute(
+      "ALTER FUNCTION realtime.broadcast_changes(text, text, text, text, text, record, record, text) OWNER TO supabase_realtime_admin"
+    )
+
+    execute(
+      "ALTER FUNCTION realtime.build_prepared_statement_sql(text, regclass, realtime.wal_column[]) OWNER TO supabase_realtime_admin"
+    )
+
+    execute("ALTER FUNCTION realtime.cast(text, regtype) OWNER TO supabase_realtime_admin")
+
+    execute(
+      "ALTER FUNCTION realtime.check_equality_op(realtime.equality_op, regtype, text, text) OWNER TO supabase_realtime_admin"
+    )
+
+    execute(
+      "ALTER FUNCTION realtime.is_visible_through_filters(realtime.wal_column[], realtime.user_defined_filter[]) OWNER TO supabase_realtime_admin"
+    )
+
+    execute("ALTER FUNCTION realtime.list_changes(name, name, integer, integer) OWNER TO supabase_realtime_admin")
+    execute("ALTER FUNCTION realtime.quote_wal2json(regclass) OWNER TO supabase_realtime_admin")
+    execute("ALTER FUNCTION realtime.send(jsonb, text, text, boolean) OWNER TO supabase_realtime_admin")
+    execute("ALTER FUNCTION realtime.subscription_check_filters() OWNER TO supabase_realtime_admin")
+    execute("ALTER FUNCTION realtime.to_regrole(text) OWNER TO supabase_realtime_admin")
+    execute("ALTER FUNCTION realtime.topic() OWNER TO supabase_realtime_admin")
+
+    execute("""
+    DO $$
+    BEGIN
+      IF (SELECT rolsuper FROM pg_roles WHERE rolname = current_user) THEN
+        REVOKE supabase_realtime_admin FROM postgres;
+      END IF;
+    END $$;
+    """)
+
+    execute("REVOKE CREATE ON SCHEMA realtime FROM postgres")
+    execute("REVOKE ALL ON realtime.schema_migrations FROM anon, authenticated, service_role, postgres")
+    execute("GRANT USAGE ON SCHEMA realtime TO postgres WITH GRANT OPTION")
+  end
+end

priv/repo/dev_seeds.exs

@@ -5,6 +5,7 @@ alias Realtime.Tenants
 
 tenant_name = "realtime-dev"
 default_db_host = "127.0.0.1"
+publication = "supabase_realtime"
 
 {:ok, tenant} =
   Repo.transaction(fn ->
@@ -26,7 +27,7 @@ default_db_host = "127.0.0.1"
             "settings" => %{
               "db_name" => System.get_env("DB_NAME", "postgres"),
               "db_host" => System.get_env("DB_HOST", default_db_host),
-              "db_user" => System.get_env("DB_USER", "supabase_admin"),
+              "db_user" => System.get_env("DB_USER", "supabase_realtime_admin"),
               "db_password" => System.get_env("DB_PASSWORD", "postgres"),
               "db_port" => System.get_env("DB_PORT", "5433"),
               "region" => "us-east-1",
@@ -43,19 +44,17 @@ default_db_host = "127.0.0.1"
   end)
 
 # Reset Tenant DB
-{:ok, settings} = Database.from_tenant(tenant, "realtime_migrations", :stop)
-settings = %{settings | max_restarts: 0, ssl: false}
-{:ok, tenant_conn} = Database.connect_db(settings)
-publication = "supabase_realtime"
+{:ok, settings} = Database.from_tenant(tenant, "realtime_seeds", :stop)
+{:ok, admin_conn} = Database.connect_db(%{settings | username: "supabase_admin", max_restarts: 0, ssl: false})
 
-Postgrex.transaction(tenant_conn, fn db_conn ->
+Postgrex.transaction(admin_conn, fn db_conn ->
   [
+    "grant usage on schema realtime to postgres, anon, authenticated, service_role",
+    "grant all on schema realtime to supabase_realtime_admin with grant option",
     "drop publication if exists #{publication}",
-    "drop table if exists public.test_tenant;",
-    "create table public.test_tenant ( id SERIAL PRIMARY KEY, details text );",
-    "grant all on table public.test_tenant to anon;",
-    "grant all on table public.test_tenant to supabase_admin;",
-    "grant all on table public.test_tenant to authenticated;",
+    "drop table if exists public.test_tenant",
+    "create table public.test_tenant ( id SERIAL PRIMARY KEY, details text )",
+    "grant all on table public.test_tenant to anon, authenticated, supabase_realtime_admin",
     "create publication #{publication} for table public.test_tenant"
   ]
   |> Enum.each(&Postgrex.query!(db_conn, &1))
@@ -66,5 +65,3 @@ case Tenants.Migrations.run_migrations(tenant) do
   :noop -> :ok
   _ -> raise "Running Migrations failed"
 end
-
-Tenants.Migrations.run_migrations(tenant)

priv/repo/seeds.exs

@@ -28,7 +28,7 @@ default_db_host = "host.docker.internal"
           "settings" => %{
             "db_name" => System.get_env("DB_NAME", "postgres"),
             "db_host" => System.get_env("DB_HOST", default_db_host),
-            "db_user" => System.get_env("DB_USER", "supabase_admin"),
+            "db_user" => System.get_env("DB_USER", "supabase_realtime_admin"),
             "db_password" => System.get_env("DB_PASSWORD", "postgres"),
             "db_port" => System.get_env("DB_PORT", "5433"),
             "region" => "us-east-1",