add worker_functions table and stopped_at column to workers

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:uE4Gy3n0TTiHs83gNYMJ63aaTl7OFO5tGkFZy6a1xD0=
+h1:IMyCv/DI0r48ffoJN/Lx1SsZLaotPIwNGXbIqw9iB7A=
 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=
@@ -18,3 +18,4 @@ h1:uE4Gy3n0TTiHs83gNYMJ63aaTl7OFO5tGkFZy6a1xD0=
 20251201105311_pgflow_temp_advisory_lock_for_compilation.sql h1:OmRtiaPYjPuq9P87Px2PH06gdKhHZ0Ro6GfjjS0G+Rs=
 20251204115929_pgflow_temp_is_local.sql h1:pjOFO6k8FCmbxp6S7U3fPImsqW81WwdLwq/UZK74BG4=
 20251204142050_pgflow_temp_ensure_flow_compiled_auto_detect.sql h1:VwqZiOcVaCahb6BZ918ioFLgwQcF/sy1TR9a4lSnVvs=
+20251204145037_pgflow_temp_worker_functions_schema.sql h1:5DJJEP0jcg7yapTe7t6FX2ypZIj92lGvJ12AX8g5fz4=