From 550a0fc492688b59579482fc5f698b48a3778235 Mon Sep 17 00:00:00 2001
From: Chris Gwilliams <517923+encima@users.noreply.github.com>
Date: Mon, 15 Sep 2025 13:59:42 +0300
Subject: [PATCH 1/2] add managed schema for subscription management

---
 ...915105031_add_subscription_mgmt_schema.sql | 100 ++++++++++++++++++
 1 file changed, 100 insertions(+)
 create mode 100644 migrations/db/migrations/20250915105031_add_subscription_mgmt_schema.sql

diff --git a/migrations/db/migrations/20250915105031_add_subscription_mgmt_schema.sql b/migrations/db/migrations/20250915105031_add_subscription_mgmt_schema.sql
new file mode 100644
index 000000000..89a9f6ba3
--- /dev/null
+++ b/migrations/db/migrations/20250915105031_add_subscription_mgmt_schema.sql
@@ -0,0 +1,100 @@
+-- migrate:up
+create schema if not exists supabase_subscription_mgmt;
+    
+create or replace function supabase_subscription_mgmt.pg_create_subscription(
+    arg_subscription_name text,
+    arg_connection_string text,
+    arg_publication_name text,
+    arg_slot_name text,
+    arg_copy_data boolean = true,
+    arg_origin text = 'any'
+)
+returns void language plpgsql
+security definer
+set search_path = pg_catalog, supabase_subscription_mgmt
+as $$
+declare
+  pg_version int;
+  create_subscription_cmd text;
+begin
+  -- get the postgresql version
+  select current_setting('server_version_num')::int into pg_version;
+
+  if pg_version < 160000 and arg_origin <> 'any' then
+    raise exception 'postgresql version must be 16 or higher to specify origin other than "any". current version: %', pg_version;
+  end if;
+
+  if arg_origin <> 'any' and arg_origin <> 'none' then
+    raise exception 'invalid origin: %. origin must be either "any" or "none".', arg_origin;
+  end if;
+
+  -- pg16 and later: include the origin parameter only if it's 'none', as its default is any
+  if pg_version >= 160000 and arg_origin = 'none' then
+      create_subscription_cmd := pg_catalog.format(
+        'create subscription %I connection %L publication %I with (slot_name=%L, create_slot=false, copy_data=%s, origin=%L)',
+        arg_subscription_name, arg_connection_string, arg_publication_name, arg_slot_name, arg_copy_data::text, arg_origin);
+ else
+      create_subscription_cmd := pg_catalog.format(
+        'create subscription %I connection %L publication %I with (slot_name=%L, create_slot=false, copy_data=%s)',
+        arg_subscription_name, arg_connection_string, arg_publication_name, arg_slot_name, arg_copy_data::text);
+  end if;
+
+  -- execute the create subscription command
+  execute create_subscription_cmd;
+end;
+$$;
+
+
+create or replace function supabase_subscription_mgmt.pg_alter_subscription_disable(
+  arg_subscription_name text
+)
+  returns void language plpgsql
+  security definer
+  set search_path = pg_catalog
+as $$
+begin
+    execute pg_catalog.format('alter subscription %I disable', arg_subscription_name);
+end;
+$$;
+
+create or replace function supabase_subscription_mgmt.pg_alter_subscription_enable(
+  arg_subscription_name text
+)
+  returns void language plpgsql
+  security definer
+  set search_path = pg_catalog, supabase_subscription_mgmt
+as $$
+begin
+    execute pg_catalog.format('alter subscription %I enable', arg_subscription_name);
+end;
+$$;
+
+create or replace function supabase_subscription_mgmt.pg_drop_subscription(
+    arg_subscription_name text
+)
+  returns void language plpgsql
+  security definer
+  set search_path = pg_catalog, supabase_subscription_mgmt
+as $$
+declare
+    l_slot_name text;
+    l_subconninfo text;
+begin
+    select subslotname, subconninfo
+        into l_slot_name, l_subconninfo
+        from pg_catalog.pg_subscription
+        where subname = arg_subscription_name;
+    if l_slot_name is null and l_subconninfo is null then
+        raise exception 'no subscription found for name: %', arg_subscription_name;
+    end if;
+    execute pg_catalog.format('alter subscription %I disable', arg_subscription_name);
+    execute pg_catalog.format('alter subscription %I set (slot_name = none)', arg_subscription_name);
+    execute pg_catalog.format('drop subscription %I', arg_subscription_name);
+end;
+$$;
+
+grant usage on schema supabase_subscription_mgmt to postgres;
+grant execute on all functions in schema supabase_subscription_mgmt to postgres;
+
+-- migrate:down
+

From 4d79888e3ba1af77e308854d8d4eca9abff7ee4f Mon Sep 17 00:00:00 2001
From: Chris Gwilliams <517923+encima@users.noreply.github.com>
Date: Tue, 16 Sep 2025 10:17:15 +0300
Subject: [PATCH 2/2] updated exists and priv check tests

---
 migrations/tests/database/exists.sql | 1 +
 migrations/tests/database/privs.sql  | 1 +
 2 files changed, 2 insertions(+)

diff --git a/migrations/tests/database/exists.sql b/migrations/tests/database/exists.sql
index bc19cd32f..c07235a26 100644
--- a/migrations/tests/database/exists.sql
+++ b/migrations/tests/database/exists.sql
@@ -7,3 +7,4 @@ SELECT has_schema('graphql');
 SELECT has_schema('graphql_public');
 SELECT has_schema('realtime');
 SELECT has_schema('storage');
+SELECT has_schema('supabase_subscription_mgmt');
diff --git a/migrations/tests/database/privs.sql b/migrations/tests/database/privs.sql
index ea4f1318a..528f508be 100644
--- a/migrations/tests/database/privs.sql
+++ b/migrations/tests/database/privs.sql
@@ -20,6 +20,7 @@ reset role;
 
 -- Verify extensions schema privileges
 SELECT schema_privs_are('extensions', 'postgres', array['CREATE', 'USAGE']);
+SELECT schema_privs_are('supabase_subscription_mgmt', 'postgres', array['USAGE']);
 SELECT schema_privs_are('extensions', 'anon', array['USAGE']);
 SELECT schema_privs_are('extensions', 'authenticated', array['USAGE']);
 SELECT schema_privs_are('extensions', 'service_role', array['USAGE']);