refactor: update worker credentials to use Supabase project ID and service role key (#522)

# Update worker management to use Supabase project credentials

This PR updates the `ensure_workers` function to use standard Supabase credentials instead of custom PgFlow credentials:

- Changes credential sources from `pgflow_service_role_key` and `pgflow_function_base_url` to `supabase_service_role_key` and `supabase_project_id`
- Constructs function URLs dynamically using the Supabase project ID: `https://{project_id}.supabase.co/functions/v1`
- Adds proper handling of empty credential strings by treating them as NULL using `nullif()`
- Adds a new test case to verify behavior when credentials exist but are empty strings
- Removes the helper function that configured vault secrets in e2e tests
- Updates migration file with improved comments about extensions

These changes make PgFlow more compatible with standard Supabase deployments by using the same credential sources that other Supabase services use.

Author: jumski

pkgs/core/schemas/0059_function_ensure_workers.sql

@@ -12,14 +12,18 @@ as $$
       select pgflow.is_local() as is_local
     ),
 
-    -- Get credentials: Vault secrets with local fallback for base_url only
+    -- Get credentials: Local mode uses hardcoded URL, production uses vault secrets
+    -- Empty strings are treated as NULL using nullif()
     credentials as (
       select
-        (select decrypted_secret from vault.decrypted_secrets where name = 'pgflow_service_role_key') as service_role_key,
-        coalesce(
-          (select decrypted_secret from vault.decrypted_secrets where name = 'pgflow_function_base_url'),
-          case when (select is_local from env) then 'http://kong:8000/functions/v1' end
-        ) as base_url
+        case
+          when (select is_local from env) then null
+          else nullif((select decrypted_secret from vault.decrypted_secrets where name = 'supabase_service_role_key'), '')
+        end as service_role_key,
+        case
+          when (select is_local from env) then 'http://kong:8000/functions/v1'
+          else (select 'https://' || nullif(decrypted_secret, '') || '.supabase.co/functions/v1' from vault.decrypted_secrets where name = 'supabase_project_id')
+        end as base_url
     ),
 
     -- Find functions that pass the debounce check
@@ -93,5 +97,6 @@ comment on function pgflow.ensure_workers() is
 In local mode: always pings all enabled functions (for fast restart after code changes).
 In production mode: only pings functions that have no alive workers.
 Respects debounce: skips functions pinged within their heartbeat_timeout_seconds window.
-Credentials: Uses Vault secrets (pgflow_service_role_key, pgflow_function_base_url) or local fallbacks.
+Credentials: Uses Vault secrets (supabase_service_role_key, supabase_project_id) or local fallbacks.
+URL is built from project_id: https://{project_id}.supabase.co/functions/v1
 Returns request_id from pg_net for each HTTP request made.';

…51208045933_pgflow_worker_management.sql …51208093149_pgflow_worker_management.sqlpkgs/core/supabase/migrations/20251208045933_pgflow_worker_management.sql renamed to pkgs/core/supabase/migrations/20251208093149_pgflow_worker_management.sql pkgs/core/supabase/migrations/20251208045933_pgflow_worker_management.sql renamed to pkgs/core/supabase/migrations/20251208093149_pgflow_worker_management.sql

@@ -1,6 +1,6 @@
--- Create extension "pg_net" (if not exists)
+-- Create extension "pg_net" if not exists (already present in Supabase)
 CREATE EXTENSION IF NOT EXISTS "pg_net" WITH SCHEMA "public";
--- Create extension "pg_cron" (if not exists)
+-- Create extension "pg_cron" if not exists (already present in Supabase)
 CREATE EXTENSION IF NOT EXISTS "pg_cron";
 -- Modify "workers" table
 ALTER TABLE "pgflow"."workers" ADD COLUMN "stopped_at" timestamptz NULL;
@@ -106,14 +106,18 @@ with
       select pgflow.is_local() as is_local
     ),
 
-    -- Get credentials: Vault secrets with local fallback for base_url only
+    -- Get credentials: Local mode uses hardcoded URL, production uses vault secrets
+    -- Empty strings are treated as NULL using nullif()
     credentials as (
       select
-        (select decrypted_secret from vault.decrypted_secrets where name = 'pgflow_service_role_key') as service_role_key,
-        coalesce(
-          (select decrypted_secret from vault.decrypted_secrets where name = 'pgflow_function_base_url'),
-          case when (select is_local from env) then 'http://kong:8000/functions/v1' end
-        ) as base_url
+        case
+          when (select is_local from env) then null
+          else nullif((select decrypted_secret from vault.decrypted_secrets where name = 'supabase_service_role_key'), '')
+        end as service_role_key,
+        case
+          when (select is_local from env) then 'http://kong:8000/functions/v1'
+          else (select 'https://' || nullif(decrypted_secret, '') || '.supabase.co/functions/v1' from vault.decrypted_secrets where name = 'supabase_project_id')
+        end as base_url
     ),
 
     -- Find functions that pass the debounce check
@@ -186,7 +190,8 @@ COMMENT ON FUNCTION "pgflow"."ensure_workers" IS 'Ensures worker functions are r
 In local mode: always pings all enabled functions (for fast restart after code changes).
 In production mode: only pings functions that have no alive workers.
 Respects debounce: skips functions pinged within their heartbeat_timeout_seconds window.
-Credentials: Uses Vault secrets (pgflow_service_role_key, pgflow_function_base_url) or local fallbacks.
+Credentials: Uses Vault secrets (supabase_service_role_key, supabase_project_id) or local fallbacks.
+URL is built from project_id: https://{project_id}.supabase.co/functions/v1
 Returns request_id from pg_net for each HTTP request made.';
 -- Create "mark_worker_stopped" function
 CREATE FUNCTION "pgflow"."mark_worker_stopped" ("worker_id" uuid) RETURNS void LANGUAGE sql AS $$

pkgs/core/supabase/migrations/atlas.sum

@@ -1,4 +1,4 @@
-h1:VgOpcmBfLwSXeC+q/2hT/Quc8P7YRzKa3x4cZ1z+ZRM=
+h1:rp08kLQLd1h/gfTMMGK0y+skRiqSfksIcbQYTe/7gXI=
 20250429164909_pgflow_initial.sql h1:I3n/tQIg5Q5nLg7RDoU3BzqHvFVjmumQxVNbXTPG15s=
 20250517072017_pgflow_fix_poll_for_tasks_to_use_separate_statement_for_polling.sql h1:wTuXuwMxVniCr3ONCpodpVWJcHktoQZIbqMZ3sUHKMY=
 20250609105135_pgflow_add_start_tasks_and_started_status.sql h1:ggGanW4Wyt8Kv6TWjnZ00/qVb3sm+/eFVDjGfT8qyPg=
@@ -12,4 +12,4 @@ h1:VgOpcmBfLwSXeC+q/2hT/Quc8P7YRzKa3x4cZ1z+ZRM=
 20251103222045_pgflow_fix_broadcast_order_and_timestamp_handling.sql h1:K/XnZpOmxfelsaNoJbR5HxhBrs/oW4aYja222h5cps4=
 20251104080523_pgflow_upgrade_pgmq_1_5_1.sql h1:Fw7zpMWnjhAHQ0qBJAprAvGl7dJMd8ExNHg8aKvkzTg=
 20251130000000_pgflow_auto_compilation.sql h1:qs+3qq1Vsyo0ETzbxDnmkVtSUa6XHkd/K9wF/3W46jM=
-20251208045933_pgflow_worker_management.sql h1:B9LcYfhZQ5hyoTmXLl6pd/kPik9EJOuPdGI/9THmrks=
+20251208093149_pgflow_worker_management.sql h1:/YCw+E63GSUiTkCx4jk4XgRApqKt+9by/NYoPsdJoIQ=

pkgs/core/supabase/tests/ensure_workers/credentials_from_vault.test.sql

@@ -6,11 +6,11 @@ select pgflow_tests.reset_db();
 -- Setup: Create Vault secrets
 select vault.create_secret(
   'test-service-role-key-from-vault',
-  'pgflow_service_role_key'
+  'supabase_service_role_key'
 );
 select vault.create_secret(
-  'http://vault-configured-url.example.com/functions/v1',
-  'pgflow_function_base_url'
+  'vaultproject123',
+  'supabase_project_id'
 );
 
 -- Setup: Register a worker function
@@ -51,10 +51,10 @@ set last_invoked_at = now() - interval '10 seconds';
 select * into temporary test3_result from pgflow.ensure_workers();
 
 select ok(
-  (select url = 'http://vault-configured-url.example.com/functions/v1/my-function'
+  (select url = 'https://vaultproject123.supabase.co/functions/v1/my-function'
    from net.http_request_queue
    where id = (select request_id from test3_result limit 1)),
-  'HTTP request URL is constructed from Vault pgflow_function_base_url'
+  'HTTP request URL is constructed from Vault supabase_project_id'
 );
 
 drop table test3_result;

pkgs/core/supabase/tests/ensure_workers/credentials_local_fallback.test.sql

@@ -4,7 +4,7 @@ select plan(4);
 select pgflow_tests.reset_db();
 
 -- Ensure no Vault secrets exist
-delete from vault.secrets where name in ('pgflow_service_role_key', 'pgflow_function_base_url');
+delete from vault.secrets where name in ('supabase_service_role_key', 'supabase_project_id');
 
 -- Setup: Register a worker function
 select pgflow.track_worker_function('my-function');

pkgs/core/supabase/tests/ensure_workers/empty_credentials_skips_invocation.test.sql

@@ -0,0 +1,72 @@
+-- Test: ensure_workers() skips invocation when credentials exist but are empty strings
+begin;
+select plan(4);
+select pgflow_tests.reset_db();
+
+-- Setup: Create Vault secrets with EMPTY strings
+select vault.create_secret('', 'supabase_service_role_key');
+select vault.create_secret('', 'supabase_project_id');
+
+-- Setup: Register a worker function
+select pgflow.track_worker_function('my-function');
+update pgflow.worker_functions
+set last_invoked_at = now() - interval '10 seconds';
+
+-- Simulate production mode (non-local jwt_secret)
+set local app.settings.jwt_secret = 'production-secret-different-from-local';
+
+-- TEST: In production mode with empty credentials, returns empty (safe failure)
+with result as (
+  select * from pgflow.ensure_workers()
+)
+select is(
+  (select count(*) from result),
+  0::bigint,
+  'Production mode with empty credentials returns empty (safe failure)'
+);
+
+-- TEST: No HTTP requests were queued (last_invoked_at unchanged)
+select ok(
+  (select last_invoked_at < now() - interval '5 seconds'
+   from pgflow.worker_functions
+   where function_name = 'my-function'),
+  'last_invoked_at not updated when credentials are empty'
+);
+
+-- TEST: Empty project_id alone should skip (even with valid service role key)
+delete from vault.secrets where name = 'supabase_service_role_key';
+select vault.create_secret('valid-service-role-key', 'supabase_service_role_key');
+
+-- Reset debounce
+update pgflow.worker_functions
+set last_invoked_at = now() - interval '10 seconds';
+
+with result as (
+  select * from pgflow.ensure_workers()
+)
+select is(
+  (select count(*) from result),
+  0::bigint,
+  'Empty project_id alone causes skip even with valid service role key'
+);
+
+-- TEST: Empty service_role_key alone should skip (even with valid project_id)
+delete from vault.secrets where name in ('supabase_service_role_key', 'supabase_project_id');
+select vault.create_secret('', 'supabase_service_role_key');
+select vault.create_secret('validproject123', 'supabase_project_id');
+
+-- Reset debounce
+update pgflow.worker_functions
+set last_invoked_at = now() - interval '10 seconds';
+
+with result as (
+  select * from pgflow.ensure_workers()
+)
+select is(
+  (select count(*) from result),
+  0::bigint,
+  'Empty service_role_key alone causes skip even with valid project_id'
+);
+
+select finish();
+rollback;

pkgs/core/supabase/tests/ensure_workers/no_credentials_skips_invocation.test.sql

@@ -4,7 +4,7 @@ select plan(2);
 select pgflow_tests.reset_db();
 
 -- Ensure no Vault secrets exist
-delete from vault.secrets where name in ('pgflow_service_role_key', 'pgflow_function_base_url');
+delete from vault.secrets where name in ('supabase_service_role_key', 'supabase_project_id');
 
 -- Setup: Register a worker function
 select pgflow.track_worker_function('my-function');

pkgs/core/supabase/tests/ensure_workers/returns_functions_to_invoke.test.sql

@@ -4,8 +4,8 @@ select plan(4);
 select pgflow_tests.reset_db();
 
 -- Setup: Create Vault secrets for production mode tests
-select vault.create_secret('test-service-role-key', 'pgflow_service_role_key');
-select vault.create_secret('http://test.example.com/functions/v1', 'pgflow_function_base_url');
+select vault.create_secret('test-service-role-key', 'supabase_service_role_key');
+select vault.create_secret('testproject123', 'supabase_project_id');
 
 -- Setup: Register two worker functions
 select pgflow.track_worker_function('function-a');

pkgs/core/supabase/tests/ensure_workers/worker_death_detection.test.sql

@@ -4,8 +4,8 @@ select plan(5);
 select pgflow_tests.reset_db();
 
 -- Setup: Create Vault secrets for production mode tests
-select vault.create_secret('test-service-role-key', 'pgflow_service_role_key');
-select vault.create_secret('http://test.example.com/functions/v1', 'pgflow_function_base_url');
+select vault.create_secret('test-service-role-key', 'supabase_service_role_key');
+select vault.create_secret('testproject123', 'supabase_project_id');
 
 -- Setup: Register a worker function
 select pgflow.track_worker_function('my-function');

pkgs/edge-worker/tests/e2e/_helpers.ts

@@ -3,7 +3,6 @@ import { delay } from '@std/async';
 import ProgressBar from 'jsr:@deno-library/progress';
 import { dim } from 'https://deno.land/std@0.224.0/fmt/colors.ts';
 import { e2eConfig } from '../config.ts';
-import type { Sql } from 'postgres';
 
 const DEBUG = Deno.env.get('DEBUG') === '1' || Deno.env.get('VERBOSE') === '1';
 
@@ -267,26 +266,6 @@ export async function startWorker(workerName: string) {
   );
 }
 
-/**
- * Configures the vault secrets needed for ensure_workers() to make HTTP requests.
- * Must be called before setupEnsureWorkersCron() in tests.
- */
-export async function configureVaultForEnsureWorkers(
-  sql: Sql<Record<string, unknown>>
-) {
-  const baseUrl = `${e2eConfig.apiUrl}/functions/v1`;
-  log(`Configuring vault with base URL: ${baseUrl}`);
-
-  // Insert or update the base URL in vault
-  await sql`
-    INSERT INTO vault.secrets (name, secret)
-    VALUES ('pgflow_function_base_url', ${baseUrl})
-    ON CONFLICT (name) DO UPDATE SET secret = EXCLUDED.secret
-  `;
-
-  log('Vault configured successfully');
-}
-
 /**
  * Monitor workers table and cron activity in background for debugging.
  * Returns an abort function to stop monitoring.