fix: remove deprecated create_realtime_partition function and update test references

- Deleted the create_realtime_partition SQL function and related comments
- Updated test scripts to no longer call create_realtime_partition
- Replaced pgflow.read_with_poll with pgmq.read_with_poll in client code
- Ensured tests focus on existing functionality without partition creation
- Minor adjustments to test setup for consistency and clarity

Author: jumski

.changeset/pgmq-version-bump.md

@@ -0,0 +1,10 @@
+---
+'@pgflow/core': minor
+'@pgflow/edge-worker': minor
+---
+
+BREAKING CHANGE: This version requires pgmq 1.5.1 or higher and will NOT work with pgmq 1.4.4.
+
+The code now depends on schema changes introduced in pgmq 1.5.0+ (specifically the headers column in message_record type). The compatibility layer that allowed pgflow to work with pgmq 1.4.4 has been removed.
+
+If you are using Supabase, pgmq 1.5.1 is included by default in recent versions. If you are self-hosting, you must upgrade pgmq to 1.5.1 or higher before upgrading pgflow.

.claude/skills/migration-management/SKILL.md

@@ -65,11 +65,20 @@ cd pkgs/core
 ./scripts/atlas-migrate-diff your_feature_name  # NO pgflow_ prefix (auto-added)
 # Creates: supabase/migrations/TIMESTAMP_pgflow_your_feature_name.sql
 
+# ⚠️ CHECK: If changing function signatures, review the migration for CREATE OR REPLACE
+# See Troubleshooting section if you get "cannot change return type" errors
+
 pnpm nx verify-migrations core  # Validates migration + checks schemas synced
 pnpm nx gen-types core          # Regenerate TypeScript types
 pnpm nx test:pgtap core         # Verify everything works
 ```
 
+**If you manually edit a migration** (e.g., to add DROP FUNCTION):
+```bash
+./scripts/atlas-migrate-hash --yes  # Regenerate checksums after manual edits
+pnpm nx verify-migrations core      # Then verify it works
+```
+
 ### Naming Conventions
 
 - snake_case (e.g., `add_root_map_support`)
@@ -180,6 +189,33 @@ git commit -m "feat: consolidate step validation and error handling"
 
 ## Troubleshooting
 
+### ⚠️ CRITICAL: Atlas Limitation with Function Signature Changes
+
+**Atlas DOES NOT detect function return type or parameter type changes!**
+
+If Atlas generates `CREATE OR REPLACE FUNCTION` and the function signature changed, the migration will fail with:
+```
+ERROR: cannot change return type of existing function
+```
+
+**Solution:** Manually edit the generated migration to add DROP first:
+```sql
+-- WRONG (Atlas generates this):
+CREATE OR REPLACE FUNCTION pgflow.my_func() RETURNS NEW_TYPE ...
+
+-- CORRECT (manually fix to):
+DROP FUNCTION IF EXISTS pgflow.my_func(param_types_here);
+CREATE FUNCTION pgflow.my_func() RETURNS NEW_TYPE ...
+```
+
+**When this happens:**
+- Changing `RETURNS SETOF type` to `RETURNS TABLE(...)`
+- Changing `RETURNS type1` to `RETURNS type2`
+- Changing parameter types
+- Adding/removing parameters
+
+**Always test migrations with `pnpm nx supabase:reset core` to catch this!**
+
 ### Migration name exists
 
 ```bash

examples/playground/supabase/migrations/20251102211603_20251102201302_pgflow_upgrade_pgmq_1_5_1.sql

@@ -0,0 +1,45 @@
+-- Modify "set_vt_batch" function
+CREATE OR REPLACE FUNCTION "pgflow"."set_vt_batch" ("queue_name" text, "msg_ids" bigint[], "vt_offsets" integer[]) RETURNS SETOF pgmq.message_record LANGUAGE plpgsql AS $$
+DECLARE
+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');
+    sql    TEXT;
+BEGIN
+    /* ---------- safety checks ---------------------------------------------------- */
+    IF msg_ids IS NULL OR vt_offsets IS NULL OR array_length(msg_ids, 1) = 0 THEN
+        RETURN;                    -- nothing to do, return empty set
+    END IF;
+
+    IF array_length(msg_ids, 1) IS DISTINCT FROM array_length(vt_offsets, 1) THEN
+        RAISE EXCEPTION
+          'msg_ids length (%) must equal vt_offsets length (%)',
+          array_length(msg_ids, 1), array_length(vt_offsets, 1);
+    END IF;
+
+    /* ---------- dynamic statement ------------------------------------------------ */
+    /* One UPDATE joins with the unnested arrays */
+    sql := format(
+        $FMT$
+        WITH input (msg_id, vt_offset) AS (
+            SELECT  unnest($1)::bigint
+                 ,  unnest($2)::int
+        )
+        UPDATE pgmq.%I q
+        SET    vt      = clock_timestamp() + make_interval(secs => input.vt_offset),
+               read_ct = read_ct     -- no change, but keeps RETURNING list aligned
+        FROM   input
+        WHERE  q.msg_id = input.msg_id
+        RETURNING q.msg_id,
+                  q.read_ct,
+                  q.enqueued_at,
+                  q.vt,
+                  q.message,
+                  q.headers
+        $FMT$,
+        qtable
+    );
+
+    RETURN QUERY EXECUTE sql USING msg_ids, vt_offsets;
+END;
+$$;
+-- Drop "read_with_poll" function
+DROP FUNCTION "pgflow"."read_with_poll";

pkgs/core/README.md

@@ -140,6 +140,7 @@ SELECT pgflow.add_step(
 #### Root Map vs Dependent Map
 
 **Root Map Steps** process the flow's input array directly:
+
 ```sql
 -- Root map: no dependencies, processes flow input
 SELECT pgflow.add_step(
@@ -156,6 +157,7 @@ SELECT pgflow.start_flow(
 ```
 
 **Dependent Map Steps** process another step's array output:
+
 ```sql
 -- Dependent map: processes the array from 'fetch_items'
 SELECT pgflow.add_step(
@@ -169,6 +171,7 @@ SELECT pgflow.add_step(
 #### Edge Cases and Special Behaviors
 
 1. **Empty Array Cascade**: When a map step receives an empty array (`[]`):
+
    - The SQL core completes it immediately without creating tasks
    - The completed map step outputs an empty array
    - Any dependent map steps also receive empty arrays and complete immediately
@@ -184,12 +187,14 @@ SELECT pgflow.add_step(
 #### Implementation Details
 
 Map steps utilize several database fields for state management:
+
 - `initial_tasks`: Number of tasks to create (NULL until array size is known)
 - `remaining_tasks`: Tracks incomplete tasks for the step
 - `task_index`: Identifies which array element each task processes
 - `step_type`: Column value 'map' triggers map behavior
 
 The aggregation process ensures:
+
 - **Order Preservation**: Task outputs maintain array element ordering
 - **NULL Handling**: NULL outputs are included in the aggregated array
 - **Atomicity**: Aggregation occurs within the same transaction as task completion
@@ -262,15 +267,17 @@ When a workflow starts:
 The Edge Worker uses a two-phase approach to retrieve and start tasks:
 
 **Phase 1 - Reserve Messages:**
+
 ```sql
-SELECT * FROM pgflow.read_with_poll(
+SELECT * FROM pgmq.read_with_poll(
   queue_name => 'analyze_website',
   vt => 60, -- visibility timeout in seconds
   qty => 5  -- maximum number of messages to fetch
 );
 ```
 
 **Phase 2 - Start Tasks:**
+
 ```sql
 SELECT * FROM pgflow.start_tasks(
   flow_slug => 'analyze_website',
@@ -379,6 +386,7 @@ Timeouts are enforced by setting the message visibility timeout to the step's ti
 The SQL Core is the DAG orchestration engine that handles dependency resolution, step state management, and task spawning. However, workflows are defined using the TypeScript Flow DSL, which compiles user intent into the SQL primitives that populate the definition tables (`flows`, `steps`, `deps`).
 
 See the [@pgflow/dsl package](../dsl/README.md) for complete documentation on:
+
 - Expressing workflows with type-safe method chaining
 - Step types (`.step()`, `.array()`, `.map()`)
 - Compilation to SQL migrations
@@ -441,6 +449,7 @@ Map step tasks receive a fundamentally different input structure than single ste
 ```
 
 This means:
+
 - Map handlers process individual elements in isolation
 - Map handlers cannot access the original flow input (`run`)
 - Map handlers cannot access other dependencies
@@ -456,8 +465,10 @@ When a step depends on a map step, it receives the aggregated array output:
 
 // A step depending on 'process_users' receives:
 {
-  "run": { /* original flow input */ },
-  "process_users": [{"name": "Alice"}, {"name": "Bob"}]  // Full array
+  "run": {
+    /* original flow input */
+  },
+  "process_users": [{ "name": "Alice" }, { "name": "Bob" }] // Full array
 }
 ```
 

pkgs/core/schemas/0080_function_read_with_poll.sql

@@ -15,7 +15,14 @@ create or replace function pgflow.set_vt_batch(
   msg_ids BIGINT [],
   vt_offsets INTEGER []
 )
-returns setof PGMQ.MESSAGE_RECORD
+returns table(
+  msg_id bigint,
+  read_ct integer,
+  enqueued_at timestamp with time zone,
+  vt timestamp with time zone,
+  message jsonb,
+  headers jsonb
+)
 language plpgsql as
 $$
 DECLARE
@@ -50,7 +57,8 @@ BEGIN
                   q.read_ct,
                   q.enqueued_at,
                   q.vt,
-                  q.message
+                  q.message,
+                  q.headers
         $FMT$,
         qtable
     );

pkgs/core/schemas/0110_function_set_vt_batch.sql

@@ -26,7 +26,7 @@ export class PgflowSqlClient<TFlow extends AnyFlow>
   ): Promise<MessageRecord[]> {
     return await this.sql<MessageRecord[]>`
       SELECT *
-      FROM pgflow.read_with_poll(
+      FROM pgmq.read_with_poll(
         queue_name => ${queueName},
         vt => ${visibilityTimeout},
         qty => ${batchSize},