Add step_index to steps and status timestamps to runtime tables (#89)

* feat: add step_index column, enforce uniqueness, auto-increment, and isolate step indexes per flow

- Added step_index column with default 0 and unique constraint within each flow
- Updated add_step function to assign step_index automatically based on max existing index
- Implemented tests for step_index auto-increment, uniqueness across flows, and flow isolation
- Enhanced flow step management to ensure step_index consistency and isolation between flows

* refactor: update pgflow migration SQL to add step_index and constraints

Renamed migration file for consistency, modified steps table to include step_index with unique
constraint, and adjusted add_step function for new schema.
Also updated checksum file accordingly.

* feat: add timestamp columns and constraints for improved auditability and data integrity

- Added created_at columns with default now() to tables: flows, steps, deps, step_states,
step_tasks
- Included completed_at, failed_at, started_at columns with relevant constraints for timing
consistency
- Updated task and step state functions to set completed_at, failed_at, and started_at timestamps
- Enhanced run status updates to record failed_at timestamps on failure
- Improved start ready steps function to timestamp started_at when transitioning steps to
'started' status

* fix(timestamps): Ensure timestamp fields are correctly set and validated in various workflows

- Corrected SQL constraint syntax in runtime schema
- Added comprehensive tests for creation, start, complete, and fail timestamps
- Included tests for timestamp fields in flow creation, step start, task completion, and failure
- Ensured timestamps are set within expected time bounds during operations
- Improved timestamp validation across multiple test scenarios for consistency and accuracy

* feat: add timestamps to migration tables and update run completion logic

- Renamed migration file for consistency
- Enhanced tables with created_at, started_at, completed_at, and failed_at timestamps
- Updated pgflow.maybe_complete_run to set completed_at and output
- Modified step and task update queries to record timestamps on status changes
- Adjusted indexes to include new timestamp columns
- Updated test SQL to verify timestamp fields are correctly set during task completion

* fix: Correct timestamp constraints and rename initial migration for consistency

- Updated timestamp constraints to ensure logical ordering in tables
- Renamed initial migration file for clarity and versioning
- Adjusted index creation statements to match updated table definitions

* fix: update verify-migrations inputs to include migrations directory

- Changed the verify-migrations task to depend on the 'migrations' input
- Ensures proper cache invalidation and accurate verification process

* fix: Correct constraints to prevent both completed_at and failed_at being non-null simultaneously

- Updated check constraints in tables to ensure only one of completed_at or failed_at can be set
- Renamed migration file for consistency and versioning
- Adjusted index creation statements for clarity and correctness
- Ensured all constraints align with intended workflow logic and data integrity standards

* fix: update failed_at logic in SQL migrations and seed script for consistency

- Corrected failed_at assignment to set timestamp only when task has exceeded max attempts
- Applied the same fix across multiple migration files to ensure consistent failure handling
- Updated seed.sql to reflect task failure behavior in testing setup
- Renamed migration files for clarity and versioning consistency
- Minor adjustment in test script to use the correct polling function for failure simulation

* fix: standardize timestamp comparisons and fix SQL syntax in test scripts

- Corrected timestamp comparison operators from '>' to '>=' for consistency
- Fixed array syntax in add_step calls to use lowercase 'array' for compatibility
- Ensured all test assertions check for timestamps being after or equal to previous ones
- Updated SQL queries for better readability and adherence to conventions
- Minor formatting adjustments across multiple test scripts to improve consistency

* fix: update migration filenames and associated checksum references

Renamed the initial pgflow migration file for consistency and updated the checksum
in the atlas.sum file to match the new filename, ensuring migration integrity.

* test: update SQL test assertions and add missing plus signs for task completion

* fix(tests): update task completion to use 'last' status and verify timestamps

- Changed 'third' to 'last' in SQL test cases for consistency
- Ensured completed_at timestamp is set and failed_at remains null after task completion
- Updated related test assertions to reflect the new status value

* feat: add step_index and status timestamps to runtime tables

- Introduced step_index to steps
- Added various status timestamp fields to runtime tables

* test: add verification for null timestamps in dependent step states

Update SQL tests to ensure that dependent steps further down the line have null values for
started_at, completed_at, and failed_at, confirming correct initial state setup in the workflow.

Author: jumski

.changeset/twenty-wasps-admire.md

@@ -0,0 +1,5 @@
+---
+'@pgflow/core': patch
+---
+
+Add step_index to steps and various status timestamps to runtime tables

pkgs/core/project.json

@@ -44,7 +44,7 @@
     "verify-migrations": {
       "executor": "nx:run-commands",
       "dependsOn": ["verify-schemas-synced", "supabase:ensure-started"],
-      "inputs": ["migrationVerificationCache"],
+      "inputs": ["migrations", "migrationVerificationCache"],
       "outputs": ["{projectRoot}/.nx-inputs/verify-migrations.txt"],
       "options": {
         "cwd": "{projectRoot}",

pkgs/core/schemas/0050_tables_definitions.sql

@@ -6,6 +6,7 @@ create table pgflow.flows (
   opt_max_attempts int not null default 3,
   opt_base_delay int not null default 1,
   opt_timeout int not null default 60,
+  created_at timestamptz not null default now(),
   constraint slug_is_valid check (pgflow.is_valid_slug(flow_slug)),
   constraint opt_max_attempts_is_nonnegative check (opt_max_attempts >= 0),
   constraint opt_base_delay_is_nonnegative check (opt_base_delay >= 0),
@@ -17,11 +18,14 @@ create table pgflow.steps (
   flow_slug text not null references pgflow.flows (flow_slug),
   step_slug text not null,
   step_type text not null default 'single',
+  step_index int not null default 0,
   deps_count int not null default 0 check (deps_count >= 0),
   opt_max_attempts int,
   opt_base_delay int,
   opt_timeout int,
+  created_at timestamptz not null default now(),
   primary key (flow_slug, step_slug),
+  unique (flow_slug, step_index),  -- Ensure step_index is unique within a flow
   check (pgflow.is_valid_slug(step_slug)),
   check (step_type in ('single')),
   constraint opt_max_attempts_is_nonnegative check (opt_max_attempts is null or opt_max_attempts >= 0),
@@ -34,6 +38,7 @@ create table pgflow.deps (
   flow_slug text not null references pgflow.flows (flow_slug),
   dep_slug text not null, -- slug of the dependency
   step_slug text not null, -- slug of the dependent
+  created_at timestamptz not null default now(),
   primary key (flow_slug, dep_slug, step_slug),
   foreign key (flow_slug, dep_slug)
   references pgflow.steps (flow_slug, step_slug),

pkgs/core/schemas/0060_tables_runtime.sql

@@ -8,7 +8,13 @@ create table pgflow.runs (
   input jsonb not null,
   output jsonb,
   remaining_steps int not null default 0 check (remaining_steps >= 0),
-  check (status in ('started', 'failed', 'completed'))
+  started_at timestamptz not null default now(),
+  completed_at timestamptz,
+  failed_at timestamptz,
+  constraint completed_at_or_failed_at check (not (completed_at is not null and failed_at is not null)),
+  constraint completed_at_is_after_started_at check (completed_at is null or completed_at >= started_at),
+  constraint failed_at_is_after_started_at check (failed_at is null or failed_at >= started_at),
+  constraint status_is_valid check (status in ('started', 'failed', 'completed'))
 );
 
 create index if not exists idx_runs_flow_slug on pgflow.runs (flow_slug);
@@ -22,11 +28,19 @@ create table pgflow.step_states (
   status text not null default 'created',
   remaining_tasks int not null default 1 check (remaining_tasks >= 0),
   remaining_deps int not null default 0 check (remaining_deps >= 0),
+  created_at timestamptz not null default now(),
+  started_at timestamptz,
+  completed_at timestamptz,
+  failed_at timestamptz,
   primary key (run_id, step_slug),
   foreign key (flow_slug, step_slug)
   references pgflow.steps (flow_slug, step_slug),
-  check (status in ('created', 'started', 'completed', 'failed')),
-  check (status != 'completed' or remaining_tasks = 0)
+  constraint status_is_valid check (status in ('created', 'started', 'completed', 'failed')),
+  constraint status_and_remaining_tasks_match check (status != 'completed' or remaining_tasks = 0),
+  constraint completed_at_or_failed_at check (not (completed_at is not null and failed_at is not null)),
+  constraint started_at_is_after_created_at check (started_at is null or started_at >= created_at),
+  constraint completed_at_is_after_started_at check (completed_at is null or completed_at >= started_at),
+  constraint failed_at_is_after_started_at check (failed_at is null or failed_at >= started_at)
 );
 
 create index if not exists idx_step_states_ready on pgflow.step_states (run_id, status, remaining_deps) where status
@@ -46,6 +60,9 @@ create table pgflow.step_tasks (
   attempts_count int not null default 0,
   error_message text,
   output jsonb,
+  queued_at timestamptz not null default now(),
+  completed_at timestamptz,
+  failed_at timestamptz,
   constraint step_tasks_pkey primary key (run_id, step_slug, task_index),
   foreign key (run_id, step_slug)
   references pgflow.step_states (run_id, step_slug),
@@ -56,7 +73,10 @@ create table pgflow.step_tasks (
     output is null or status = 'completed'
   ),
   constraint only_single_task_per_step check (task_index = 0),
-  constraint attempts_count_nonnegative check (attempts_count >= 0)
+  constraint attempts_count_nonnegative check (attempts_count >= 0),
+  constraint completed_at_or_failed_at check (not (completed_at is not null and failed_at is not null)),
+  constraint completed_at_is_after_queued_at check (completed_at is null or completed_at >= queued_at),
+  constraint failed_at_is_after_queued_at check (failed_at is null or failed_at >= queued_at)
 );
 
 create index if not exists idx_step_tasks_message_id on pgflow.step_tasks (message_id);

pkgs/core/schemas/0100_function_add_step.sql

@@ -12,9 +12,15 @@ set search_path to ''
 volatile
 as $$
 WITH
+  next_index AS (
+    SELECT COALESCE(MAX(step_index) + 1, 0) as idx
+    FROM pgflow.steps
+    WHERE flow_slug = add_step.flow_slug
+  ),
   create_step AS (
-    INSERT INTO pgflow.steps (flow_slug, step_slug, deps_count, opt_max_attempts, opt_base_delay, opt_timeout)
-    VALUES (flow_slug, step_slug, COALESCE(array_length(deps_slugs, 1), 0), max_attempts, base_delay, timeout)
+    INSERT INTO pgflow.steps (flow_slug, step_slug, step_index, deps_count, opt_max_attempts, opt_base_delay, opt_timeout)
+    SELECT add_step.flow_slug, add_step.step_slug, idx, COALESCE(array_length(deps_slugs, 1), 0), max_attempts, base_delay, timeout
+    FROM next_index
     ON CONFLICT (flow_slug, step_slug)
     DO UPDATE SET step_slug = pgflow.steps.step_slug
     RETURNING *

pkgs/core/schemas/0100_function_complete_task.sql

@@ -26,6 +26,7 @@ task AS (
   UPDATE pgflow.step_tasks
   SET
     status = 'completed',
+    completed_at = now(),
     output = complete_task.output
   WHERE pgflow.step_tasks.run_id = complete_task.run_id
     AND pgflow.step_tasks.step_slug = complete_task.step_slug
@@ -39,6 +40,10 @@ step_state AS (
     WHEN pgflow.step_states.remaining_tasks = 1 THEN 'completed'  -- Will be 0 after decrement
     ELSE 'started'
     END,
+    completed_at = CASE
+    WHEN pgflow.step_states.remaining_tasks = 1 THEN now()  -- Will be 0 after decrement
+    ELSE NULL
+    END,
     remaining_tasks = pgflow.step_states.remaining_tasks - 1
   FROM task
   WHERE pgflow.step_states.run_id = complete_task.run_id

pkgs/core/schemas/0100_function_fail_task.sql

@@ -44,6 +44,10 @@ fail_or_retry_task as (
       WHEN task.attempts_count < (SELECT opt_max_attempts FROM config) THEN 'queued'
       ELSE 'failed'
     END,
+    failed_at = CASE
+      WHEN task.attempts_count >= (SELECT opt_max_attempts FROM config) THEN now()
+      ELSE NULL
+    END,
     error_message = fail_task.error_message
   WHERE task.run_id = fail_task.run_id
     AND task.step_slug = fail_task.step_slug
@@ -57,7 +61,11 @@ maybe_fail_step AS (
     status = CASE
              WHEN (select fail_or_retry_task.status from fail_or_retry_task) = 'failed' THEN 'failed'
              ELSE pgflow.step_states.status
-             END
+             END,
+    failed_at = CASE
+                WHEN (select fail_or_retry_task.status from fail_or_retry_task) = 'failed' THEN now()
+                ELSE NULL
+                END
   FROM fail_or_retry_task
   WHERE pgflow.step_states.run_id = fail_task.run_id
     AND pgflow.step_states.step_slug = fail_task.step_slug
@@ -67,7 +75,11 @@ UPDATE pgflow.runs
 SET status = CASE
               WHEN (select status from maybe_fail_step) = 'failed' THEN 'failed'
               ELSE status
-              END
+              END,
+    failed_at = CASE
+                WHEN (select status from maybe_fail_step) = 'failed' THEN now()
+                ELSE NULL
+                END
 WHERE pgflow.runs.run_id = fail_task.run_id;
 
 -- For queued tasks: delay the message for retry with exponential backoff

pkgs/core/schemas/0100_function_maybe_complete_run.sql

@@ -9,6 +9,7 @@ as $$
   UPDATE pgflow.runs
   SET
     status = 'completed',
+    completed_at = now(),
     output = (
       -- Get outputs from final steps (steps that are not dependencies for other steps)
       SELECT jsonb_object_agg(st.step_slug, st.output)

pkgs/core/schemas/0100_function_start_ready_steps.sql

@@ -15,7 +15,8 @@ WITH ready_steps AS (
 ),
 started_step_states AS (
   UPDATE pgflow.step_states
-  SET status = 'started'
+  SET status = 'started',
+      started_at = now()
   FROM ready_steps
   WHERE pgflow.step_states.run_id = start_ready_steps.run_id
     AND pgflow.step_states.step_slug = ready_steps.step_slug