add setup_ensure_workers_cron() SQL function for cron job scheduling

Author: jumski

pkgs/core/schemas/0061_function_setup_ensure_workers_cron.sql

@@ -0,0 +1,59 @@
+-- Setup Ensure Workers Cron
+-- Sets up cron jobs for worker management: ensure_workers (configurable interval) and cleanup_logs (hourly)
+
+drop function if exists pgflow.setup_ensure_workers_cron(text);
+
+create or replace function pgflow.setup_ensure_workers_cron(
+  cron_interval text default '1 second'
+)
+returns text
+language plpgsql
+security definer
+set search_path = pgflow, cron, pg_temp
+as $$
+declare
+  ensure_workers_job_id bigint;
+  cleanup_job_id bigint;
+begin
+  -- Remove existing jobs if they exist (ignore errors if not found)
+  begin
+    perform cron.unschedule('pgflow_ensure_workers');
+  exception when others then
+    -- Job doesn't exist, continue
+  end;
+
+  begin
+    perform cron.unschedule('pgflow_cleanup_logs');
+  exception when others then
+    -- Job doesn't exist, continue
+  end;
+
+  -- Schedule ensure_workers job with the specified interval
+  ensure_workers_job_id := cron.schedule(
+    job_name => 'pgflow_ensure_workers',
+    schedule => setup_ensure_workers_cron.cron_interval,
+    command => 'select pgflow.ensure_workers()'
+  );
+
+  -- Schedule cleanup job to run hourly
+  cleanup_job_id := cron.schedule(
+    job_name => 'pgflow_cleanup_logs',
+    schedule => '0 * * * *',
+    command => 'select pgflow.cleanup_ensure_workers_logs()'
+  );
+
+  return format(
+    'Scheduled pgflow_ensure_workers (every %s, job_id=%s) and pgflow_cleanup_logs (hourly, job_id=%s)',
+    setup_ensure_workers_cron.cron_interval,
+    ensure_workers_job_id,
+    cleanup_job_id
+  );
+end;
+$$;
+
+comment on function pgflow.setup_ensure_workers_cron(text) is
+'Sets up cron jobs for worker management.
+Schedules pgflow_ensure_workers at the specified cron_interval (default: 1 second) to keep workers running.
+Schedules pgflow_cleanup_logs hourly to clean up old cron job logs.
+Replaces existing jobs if they exist (idempotent).
+Returns a confirmation message with job IDs.';

pkgs/core/src/database-types.ts

@@ -558,6 +558,10 @@ export type Database = {
           vt: string
         }[]
       }
+      setup_ensure_workers_cron: {
+        Args: { cron_interval?: string }
+        Returns: string
+      }
       start_flow: {
         Args: { flow_slug: string; input: Json; run_id?: string }
         Returns: {

pkgs/core/supabase/migrations/20251205162618_pgflow_temp_setup_ensure_workers_cron.sql

@@ -0,0 +1,47 @@
+-- Create "setup_ensure_workers_cron" function
+CREATE FUNCTION "pgflow"."setup_ensure_workers_cron" ("cron_interval" text DEFAULT '1 second') RETURNS text LANGUAGE plpgsql SECURITY DEFINER SET "search_path" = pgflow, cron, pg_temp AS $$
+declare
+  ensure_workers_job_id bigint;
+  cleanup_job_id bigint;
+begin
+  -- Remove existing jobs if they exist (ignore errors if not found)
+  begin
+    perform cron.unschedule('pgflow_ensure_workers');
+  exception when others then
+    -- Job doesn't exist, continue
+  end;
+
+  begin
+    perform cron.unschedule('pgflow_cleanup_logs');
+  exception when others then
+    -- Job doesn't exist, continue
+  end;
+
+  -- Schedule ensure_workers job with the specified interval
+  ensure_workers_job_id := cron.schedule(
+    job_name => 'pgflow_ensure_workers',
+    schedule => setup_ensure_workers_cron.cron_interval,
+    command => 'select pgflow.ensure_workers()'
+  );
+
+  -- Schedule cleanup job to run hourly
+  cleanup_job_id := cron.schedule(
+    job_name => 'pgflow_cleanup_logs',
+    schedule => '0 * * * *',
+    command => 'select pgflow.cleanup_ensure_workers_logs()'
+  );
+
+  return format(
+    'Scheduled pgflow_ensure_workers (every %s, job_id=%s) and pgflow_cleanup_logs (hourly, job_id=%s)',
+    setup_ensure_workers_cron.cron_interval,
+    ensure_workers_job_id,
+    cleanup_job_id
+  );
+end;
+$$;
+-- Set comment to function: "setup_ensure_workers_cron"
+COMMENT ON FUNCTION "pgflow"."setup_ensure_workers_cron" IS 'Sets up cron jobs for worker management.
+Schedules pgflow_ensure_workers at the specified cron_interval (default: 1 second) to keep workers running.
+Schedules pgflow_cleanup_logs hourly to clean up old cron job logs.
+Replaces existing jobs if they exist (idempotent).
+Returns a confirmation message with job IDs.';

pkgs/core/supabase/migrations/atlas.sum

@@ -1,4 +1,4 @@
-h1:UPs/YAMNWH5FdnHa1GJRQZ4yiv6kOHpfBNfs6telwgI=
+h1:ptkKThTquu9rJzpsRUMDDHbe4OgnCEYGATLSRy0F1l0=
 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=
@@ -20,3 +20,4 @@ h1:UPs/YAMNWH5FdnHa1GJRQZ4yiv6kOHpfBNfs6telwgI=
 20251205103442_pgflow_temp_add_extensions.sql h1:IBHG1vBdXu8wDEJzqpJUFmuPhVaX0mAmDUkngLgdaMg=
 20251205133446_pgflow_temp_ensure_workers.sql h1:EQzE75uaMSXeU1sdjO7MK1ipCwepxlWSVzlKegLpr48=
 20251205140756_pgflow_temp_cleanup_ensure_workers_logs.sql h1:wNsRFg6bZKwItg3qdNx/dP8zaK4Qe7ufOZwxO+yx/F8=
+20251205162618_pgflow_temp_setup_ensure_workers_cron.sql h1:Kq80xDBLQc60yYlrntZEFefQ4kfeiycMbauU7LKljCY=

pkgs/core/supabase/tests/setup_ensure_workers_cron/custom_interval.test.sql

@@ -0,0 +1,22 @@
+-- Test: setup_ensure_workers_cron() accepts custom interval
+begin;
+select plan(2);
+
+-- Execute function with custom interval (handles existing jobs internally)
+select pgflow.setup_ensure_workers_cron(cron_interval => '5 seconds');
+
+-- Test: Job is created with custom interval
+select ok(
+  exists(select 1 from cron.job where jobname = 'pgflow_ensure_workers'),
+  'Should create pgflow_ensure_workers job'
+);
+
+-- Test: Job has the custom schedule
+select is(
+  (select schedule from cron.job where jobname = 'pgflow_ensure_workers'),
+  '5 seconds',
+  'Job should have 5 seconds schedule as specified'
+);
+
+select finish();
+rollback;

pkgs/core/supabase/tests/setup_ensure_workers_cron/replaces_existing_job.test.sql

@@ -0,0 +1,33 @@
+-- Test: setup_ensure_workers_cron() replaces existing job when called twice
+begin;
+select plan(3);
+
+-- Setup first job with 1 second interval
+select pgflow.setup_ensure_workers_cron(cron_interval => '1 second');
+
+-- Get job count after first call
+select is(
+  (select count(*)::int from cron.job where jobname = 'pgflow_ensure_workers'),
+  1,
+  'Should have exactly one ensure_workers job after first call'
+);
+
+-- Call again with different interval (should replace, not duplicate)
+select pgflow.setup_ensure_workers_cron(cron_interval => '2 seconds');
+
+-- Test: Still only one job (no duplicates)
+select is(
+  (select count(*)::int from cron.job where jobname = 'pgflow_ensure_workers'),
+  1,
+  'Should still have exactly one ensure_workers job after second call'
+);
+
+-- Test: Job has updated schedule
+select is(
+  (select schedule from cron.job where jobname = 'pgflow_ensure_workers'),
+  '2 seconds',
+  'Job should have updated schedule after second call'
+);
+
+select finish();
+rollback;

pkgs/core/supabase/tests/setup_ensure_workers_cron/returns_confirmation.test.sql

@@ -0,0 +1,22 @@
+-- Test: setup_ensure_workers_cron() returns confirmation message
+begin;
+select plan(2);
+
+-- Execute function and capture result (handles existing jobs internally)
+select pgflow.setup_ensure_workers_cron() as message into temporary result;
+
+-- Test: Returns a non-null message
+select ok(
+  (select message from result) is not null,
+  'Should return a confirmation message'
+);
+
+-- Test: Message mentions the scheduled jobs
+select ok(
+  (select message from result) like '%pgflow_ensure_workers%',
+  'Confirmation message should mention the scheduled job'
+);
+
+drop table result;
+select finish();
+rollback;

pkgs/core/supabase/tests/setup_ensure_workers_cron/schedules_cleanup_job.test.sql

@@ -0,0 +1,28 @@
+-- Test: setup_ensure_workers_cron() schedules the cleanup job
+begin;
+select plan(3);
+
+-- Execute function (should be idempotent - handles existing jobs internally)
+select pgflow.setup_ensure_workers_cron();
+
+-- Test: Cleanup job is created
+select ok(
+  exists(select 1 from cron.job where jobname = 'pgflow_cleanup_logs'),
+  'Should create pgflow_cleanup_logs job'
+);
+
+-- Test: Cleanup job runs hourly
+select is(
+  (select schedule from cron.job where jobname = 'pgflow_cleanup_logs'),
+  '0 * * * *',
+  'Cleanup job should run hourly (every hour at minute 0)'
+);
+
+-- Test: Cleanup job runs the correct command
+select ok(
+  (select command from cron.job where jobname = 'pgflow_cleanup_logs') like '%pgflow.cleanup_ensure_workers_logs()%',
+  'Cleanup job should call pgflow.cleanup_ensure_workers_logs()'
+);
+
+select finish();
+rollback;

pkgs/core/supabase/tests/setup_ensure_workers_cron/schedules_ensure_workers_job.test.sql

@@ -0,0 +1,28 @@
+-- Test: setup_ensure_workers_cron() schedules the ensure_workers job
+begin;
+select plan(3);
+
+-- Execute function (should be idempotent - handles existing jobs internally)
+select pgflow.setup_ensure_workers_cron();
+
+-- Test: Job is created
+select ok(
+  exists(select 1 from cron.job where jobname = 'pgflow_ensure_workers'),
+  'Should create pgflow_ensure_workers job'
+);
+
+-- Test: Job has correct schedule (1 second default)
+select is(
+  (select schedule from cron.job where jobname = 'pgflow_ensure_workers'),
+  '1 second',
+  'Job should have 1 second schedule by default'
+);
+
+-- Test: Job runs the correct command
+select ok(
+  (select command from cron.job where jobname = 'pgflow_ensure_workers') like '%pgflow.ensure_workers()%',
+  'Job should call pgflow.ensure_workers()'
+);
+
+select finish();
+rollback;