feat: add track_worker_function for edge function monitoring (#505)

# Add track_worker_function for edge function monitoring

This PR introduces a new SQL function `pgflow.track_worker_function()` that allows edge functions to register themselves for monitoring by the `ensure_workers()` cron job.

The function:
- Takes a function name as input
- Inserts or updates a record in the `pgflow.worker_functions` table
- Updates the `updated_at` timestamp on each call

This enables worker functions to "check in" periodically, allowing the system to detect and restart any workers that have stopped responding.

Added comprehensive tests to verify:
- New worker functions are inserted correctly
- Default values are set appropriately (enabled=true, heartbeat_timeout=6)
- Timestamps are updated properly on subsequent calls
- Multiple different functions can be tracked

Also updated TypeScript database types to include the new function.

Author: jumski

pkgs/core/schemas/0057_function_track_worker_function.sql

@@ -0,0 +1,21 @@
+-- Track Worker Function
+-- Registers an edge function for monitoring by ensure_workers() cron
+
+drop function if exists pgflow.track_worker_function(text);
+
+create or replace function pgflow.track_worker_function(
+  function_name text
+) returns void
+language sql
+as $$
+  insert into pgflow.worker_functions (function_name, updated_at, last_invoked_at)
+  values (track_worker_function.function_name, clock_timestamp(), clock_timestamp())
+  on conflict (function_name)
+  do update set
+    updated_at = clock_timestamp(),
+    last_invoked_at = clock_timestamp();
+$$;
+
+comment on function pgflow.track_worker_function(text) is
+'Registers an edge function for monitoring. Called by workers on startup.
+Sets last_invoked_at to prevent cron from pinging during startup (debounce).';

pkgs/core/src/database-types.ts

@@ -578,6 +578,10 @@ export type Database = {
           isSetofReturn: true
         }
       }
+      track_worker_function: {
+        Args: { function_name: string }
+        Returns: undefined
+      }
     }
     Enums: {
       [_ in never]: never

pkgs/core/supabase/migrations/20251204164612_pgflow_temp_track_worker_function.sql

@@ -0,0 +1,12 @@
+-- Create "track_worker_function" function
+CREATE FUNCTION "pgflow"."track_worker_function" ("function_name" text) RETURNS void LANGUAGE sql AS $$
+insert into pgflow.worker_functions (function_name, updated_at, last_invoked_at)
+  values (track_worker_function.function_name, clock_timestamp(), clock_timestamp())
+  on conflict (function_name)
+  do update set
+    updated_at = clock_timestamp(),
+    last_invoked_at = clock_timestamp();
+$$;
+-- Set comment to function: "track_worker_function"
+COMMENT ON FUNCTION "pgflow"."track_worker_function" IS 'Registers an edge function for monitoring. Called by workers on startup.
+Sets last_invoked_at to prevent cron from pinging during startup (debounce).';

pkgs/core/supabase/migrations/atlas.sum

@@ -1,4 +1,4 @@
-h1:1+E3ra0bCEMcyTaOltxInks1QXaP9l45RgA3GSyY7To=
+h1:Uh2fD3kgXbd4Emzw0ZL3tDr0KKpzWSFzI0tkl0XnytM=
 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=
@@ -15,3 +15,4 @@ h1:1+E3ra0bCEMcyTaOltxInks1QXaP9l45RgA3GSyY7To=
 20251204115929_pgflow_temp_is_local.sql h1:arP+PC2OYI9ktAFx0+G9/w8zsaq/AbFWjLgK6YDujvQ=
 20251204142050_pgflow_temp_ensure_flow_compiled_auto_detect.sql h1:9FsEd0iyaIv9X1alACBQCzyebt2/+m1rgL4ozXJWBcA=
 20251204145037_pgflow_temp_worker_functions_schema.sql h1:mrLKtkM8aWHBY+LIxQrsXxE7aKuPJaQJO2qd0NnrJ74=
+20251204164612_pgflow_temp_track_worker_function.sql h1:3Ht8wUx3saKPo98osuTG/nxD/tKR48qe8jHNVwuu2lY=

pkgs/core/supabase/tests/track_worker_function/basic.test.sql

@@ -0,0 +1,75 @@
+begin;
+select plan(9);
+select pgflow_tests.reset_db();
+
+-- TEST: Inserts new worker function when it does not exist
+select pgflow.track_worker_function('my-edge-function');
+select is(
+  (select count(*) from pgflow.worker_functions where function_name = 'my-edge-function'),
+  1::bigint,
+  'track_worker_function() inserts new worker function'
+);
+
+-- TEST: New worker function has enabled=true by default
+select is(
+  (select enabled from pgflow.worker_functions where function_name = 'my-edge-function'),
+  true,
+  'New worker function has enabled=true by default'
+);
+
+-- TEST: New worker function has default heartbeat_timeout_seconds
+select is(
+  (select heartbeat_timeout_seconds from pgflow.worker_functions where function_name = 'my-edge-function'),
+  6,
+  'New worker function has heartbeat_timeout_seconds=6 by default'
+);
+
+-- TEST: New worker function has last_invoked_at set (debounce protection)
+select isnt(
+  (select last_invoked_at from pgflow.worker_functions where function_name = 'my-edge-function'),
+  null::timestamptz,
+  'New worker function has last_invoked_at set on insert (debounce protection)'
+);
+
+-- TEST: last_invoked_at is set to approximately now
+select ok(
+  (select last_invoked_at >= now() - interval '1 second'
+   from pgflow.worker_functions
+   where function_name = 'my-edge-function'),
+  'last_invoked_at is set to approximately now on insert'
+);
+
+-- TEST: Upsert updates updated_at on conflict
+-- First, get the original updated_at timestamp
+select pg_sleep(0.01); -- Small delay to ensure timestamp difference
+select pgflow.track_worker_function('my-edge-function');
+select ok(
+  (select updated_at > created_at from pgflow.worker_functions where function_name = 'my-edge-function'),
+  'Upsert updates updated_at timestamp on conflict'
+);
+
+-- TEST: Upsert updates last_invoked_at on conflict
+select ok(
+  (select last_invoked_at >= now() - interval '1 second'
+   from pgflow.worker_functions
+   where function_name = 'my-edge-function'),
+  'Upsert updates last_invoked_at on conflict (refreshes debounce)'
+);
+
+-- TEST: Upsert does not duplicate rows
+select is(
+  (select count(*) from pgflow.worker_functions where function_name = 'my-edge-function'),
+  1::bigint,
+  'Upsert does not create duplicate rows'
+);
+
+-- TEST: Can track multiple different functions
+select pgflow.track_worker_function('another-function');
+select is(
+  (select count(*) from pgflow.worker_functions),
+  2::bigint,
+  'Can track multiple different worker functions'
+);
+
+select finish();
+rollback;