feat: add worker_functions table and stopped_at column to workers (#504)

# Add worker_functions table and stopped_at column to workers

This PR introduces a new `worker_functions` table to track which edge functions should be pinged by the `ensure_workers()` cron job. The table includes configuration options like:

- `enabled` flag to control whether a function should be pinged
- `heartbeat_timeout_seconds` to define when a worker is considered dead
- Timestamps for creation and updates

Additionally, adds a `stopped_at` column to the existing `workers` table to properly track when workers are stopped.

The changes include:

- Schema definitions in core schemas
- Migration script for Supabase
- Updated TypeScript type definitions

Author: jumski

pkgs/core/schemas/0055_tables_workers.sql

@@ -6,6 +6,7 @@ create table if not exists pgflow.workers (
   function_name text not null,
   started_at timestamptz not null default now(),
   deprecated_at timestamptz,
+  stopped_at timestamptz,
   last_heartbeat_at timestamptz not null default now()
 );
 

pkgs/core/schemas/0056_table_worker_functions.sql

@@ -0,0 +1,26 @@
+-- Worker Functions Table
+-- Tracks which edge functions should be pinged by ensure_workers() cron
+
+create table if not exists pgflow.worker_functions (
+  function_name text not null primary key,
+  enabled boolean not null default true,
+  heartbeat_timeout_seconds integer not null default 6,
+  last_invoked_at timestamptz,
+  created_at timestamptz not null default now(),
+  updated_at timestamptz not null default now()
+);
+
+comment on table pgflow.worker_functions is
+'Registry of edge functions that run pgflow workers, used by ensure_workers() cron';
+
+comment on column pgflow.worker_functions.function_name is
+'Name of the Supabase Edge Function';
+
+comment on column pgflow.worker_functions.enabled is
+'Whether ensure_workers() should ping this function';
+
+comment on column pgflow.worker_functions.heartbeat_timeout_seconds is
+'How long before considering a worker dead (no heartbeat)';
+
+comment on column pgflow.worker_functions.last_invoked_at is
+'When ensure_workers() last pinged this function (used for debouncing)';

pkgs/core/src/database-types.ts

@@ -314,13 +314,41 @@ export type Database = {
           },
         ]
       }
+      worker_functions: {
+        Row: {
+          created_at: string
+          enabled: boolean
+          function_name: string
+          heartbeat_timeout_seconds: number
+          last_invoked_at: string | null
+          updated_at: string
+        }
+        Insert: {
+          created_at?: string
+          enabled?: boolean
+          function_name: string
+          heartbeat_timeout_seconds?: number
+          last_invoked_at?: string | null
+          updated_at?: string
+        }
+        Update: {
+          created_at?: string
+          enabled?: boolean
+          function_name?: string
+          heartbeat_timeout_seconds?: number
+          last_invoked_at?: string | null
+          updated_at?: string
+        }
+        Relationships: []
+      }
       workers: {
         Row: {
           deprecated_at: string | null
           function_name: string
           last_heartbeat_at: string
           queue_name: string
           started_at: string
+          stopped_at: string | null
           worker_id: string
         }
         Insert: {
@@ -329,6 +357,7 @@ export type Database = {
           last_heartbeat_at?: string
           queue_name: string
           started_at?: string
+          stopped_at?: string | null
           worker_id: string
         }
         Update: {
@@ -337,6 +366,7 @@ export type Database = {
           last_heartbeat_at?: string
           queue_name?: string
           started_at?: string
+          stopped_at?: string | null
           worker_id?: string
         }
         Relationships: []

pkgs/core/supabase/migrations/20251204145037_pgflow_temp_worker_functions_schema.sql

@@ -0,0 +1,22 @@
+-- Modify "workers" table
+ALTER TABLE "pgflow"."workers" ADD COLUMN "stopped_at" timestamptz NULL;
+-- Create "worker_functions" table
+CREATE TABLE "pgflow"."worker_functions" (
+  "function_name" text NOT NULL,
+  "enabled" boolean NOT NULL DEFAULT true,
+  "heartbeat_timeout_seconds" integer NOT NULL DEFAULT 6,
+  "last_invoked_at" timestamptz NULL,
+  "created_at" timestamptz NOT NULL DEFAULT now(),
+  "updated_at" timestamptz NOT NULL DEFAULT now(),
+  PRIMARY KEY ("function_name")
+);
+-- Set comment to table: "worker_functions"
+COMMENT ON TABLE "pgflow"."worker_functions" IS 'Registry of edge functions that run pgflow workers, used by ensure_workers() cron';
+-- Set comment to column: "function_name" on table: "worker_functions"
+COMMENT ON COLUMN "pgflow"."worker_functions"."function_name" IS 'Name of the Supabase Edge Function';
+-- Set comment to column: "enabled" on table: "worker_functions"
+COMMENT ON COLUMN "pgflow"."worker_functions"."enabled" IS 'Whether ensure_workers() should ping this function';
+-- Set comment to column: "heartbeat_timeout_seconds" on table: "worker_functions"
+COMMENT ON COLUMN "pgflow"."worker_functions"."heartbeat_timeout_seconds" IS 'How long before considering a worker dead (no heartbeat)';
+-- Set comment to column: "last_invoked_at" on table: "worker_functions"
+COMMENT ON COLUMN "pgflow"."worker_functions"."last_invoked_at" IS 'When ensure_workers() last pinged this function (used for debouncing)';

pkgs/core/supabase/migrations/atlas.sum

@@ -1,4 +1,4 @@
-h1:moHUa6OxOaPfYpNNrLRhkw13YzwraGNDxMnA9aarP30=
+h1:1+E3ra0bCEMcyTaOltxInks1QXaP9l45RgA3GSyY7To=
 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=
@@ -14,3 +14,4 @@ h1:moHUa6OxOaPfYpNNrLRhkw13YzwraGNDxMnA9aarP30=
 20251130000000_pgflow_auto_compilation.sql h1:qs+3qq1Vsyo0ETzbxDnmkVtSUa6XHkd/K9wF/3W46jM=
 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=