Add way to prune old records (#118)

* feat: add SQL function and tests for pruning old records from pgflow tables

Introduce a stored procedure to clean up old flow-related records based on retention days.
Add comprehensive test suite to verify pruning logic, timestamps updates, and record counts.

* feat: add and update SQL scripts for pruning old records in pgflow

- Removed unnecessary file references from SQL comments
- Updated test SQL to reflect correct worker counts and record states
- Enhanced pruning function with comprehensive deletion logic for multiple tables
- Added documentation for scheduling and manual execution of pruning
- Included performance tip for large tables to run during off-peak hours
- Created new documentation page explaining how to prune old records and schedule maintenance
routines

* feat: update documentation and embed SQL pruning function

- Added import for Code component in documentation
- Embedded SQL code for prune_old_records function into the documentation
- Updated instructions and comments within the SQL function
- Included a performance tip for large tables

These changes enhance the documentation clarity and provide the full SQL implementation for
record pruning functionality.

* refactor: update prune_old_records function to simplify deletion logic and improve testing

- Replaced complex CTE-based deletion with direct DELETE statements for each table
- Removed redundant return values, now the function returns void
- Enhanced test SQL to better simulate old and recent records, including timestamp updates
- Added tests to verify pruning behavior, especially for still-running flows
- Improved clarity and maintainability of the pruning procedure and associated tests

* feat: extend pruning function to include archived messages from PGMQ tables

- Prune old records from pgflow tables including workers, step_tasks, step_states, and runs
- Add dynamic deletion of archived messages from pgmq archive tables based on flow slugs
- Implement dynamic SQL to handle variable archive table names and ensure proper cleanup

* chore(tests): add comprehensive tests for pruning functions and timestamp updates

Introduce new SQL test files for pruning archived messages, completed flows, failed flows,
and workers.
Add tests that verify pruning logic based on timestamps, flow states, and flow statuses.
Implement setup, execution, and validation steps to ensure correct pruning behavior across
scenarios.

* test: update pruning test to account for additional worker and adjust expectations

Enhance the prune_workers.test.sql to include a new worker at the exact cutoff time,
modify the initial worker count, and update assertions to reflect the correct number
of remaining workers after pruning.
Also, add a check to ensure the worker at the
cutoff is not pruned.
These changes improve test coverage for boundary conditions.

* chore: update test helpers, add SQL path, and improve migration concatenation

- Added sql_paths for test_helpers.sql in config.toml
- Renamed seed.sql to test_helpers.sql and updated related references
- Enhanced test helper functions for setting flow timestamps in SQL scripts
- Updated test setup scripts to use raw SQL files for pruning tests
- Fixed import path in documentation to load correct SQL file
- Modified migration concatenation script to include test_helpers.sql
- Improved test SQL files with consistent formatting and comments

* fix(supabase): correct SQL function for pruning old records and archive tables

Refactors the prune_old_records function to fix syntax issues, improve table existence checks,
and ensure proper deletion of outdated data across multiple related tables and archive tables.
Also updates the default retention period parameter.

* feat: add documentation and implementation options for pruning old records

Update the docs to clarify the purpose and usage of the pgflow.prune_old_records function,
including examples and notes on its current status.
Introduce scheduling options with pg_cron
and manual execution steps.
Include the SQL function code as a tip and add performance advice.

* refactor: rename prune_old_records to prune_data_older_than in tests and documentation

Update all test scripts, SQL imports, and documentation references to use the new function name
for clarity and consistency.

* chore: update pruning function to accept interval parameter and improve documentation

- Modified pgflow.prune_data_older_than to use an INTERVAL parameter instead of days
- Updated related SQL functions and migration snippets for flexible data retention
- Enhanced documentation with examples and scheduling instructions for automated pruning
- Adjusted test scripts to use make_interval for better clarity and flexibility
- Added notes on performance considerations and manual setup guidance

* fix(tests): update interval parameter to use make_interval in pruning tests

- Changed calls to prune_data_older_than from passing a numeric value to using make_interval(days
=> 30 or 7)
- Ensured consistency across multiple test files for pruning functions
- Preserved rollback commands and test structure without modifications

Author: jumski

pkgs/core/supabase/config.toml

@@ -8,6 +8,9 @@ major_version = 15
 # disable unused features
 [db.seed]
 enabled = true
+sql_paths = [
+  "test_helpers.sql",
+]
 
 [api]
 port = 50421  # Add custom port for Kong

pkgs/core/supabase/seed.sql renamed to pkgs/core/supabase/test_helpers.sql

@@ -200,3 +200,107 @@ BEGIN
   );
 END;
 $$ language plpgsql;
+
+/**
+ * Helper functions for pruning tests.
+ * Contains basic setup code that would otherwise be repeated.
+ */
+
+-- Set timestamps for completed runs/steps/tasks to be older than the cutoff
+create or replace function pgflow_tests.set_completed_flow_timestamps(
+  p_flow_slug text,
+  days_old integer
+) returns void language plpgsql as $$
+begin
+  -- Set timestamps for step_tasks
+  update pgflow.step_tasks
+  set
+    queued_at = now() - interval '1 day' - (days_old * interval '1 day'),
+    completed_at = now() - (days_old * interval '1 day'),
+    status = 'completed'
+  where flow_slug = p_flow_slug;
+
+  -- Set timestamps for step_states
+  update pgflow.step_states
+  set
+    created_at = now() - interval '2 days' - (days_old * interval '1 day'),
+    started_at = now() - interval '1 day' - (days_old * interval '1 day'),
+    completed_at = now() - (days_old * interval '1 day'),
+    status = 'completed',
+    remaining_tasks = 0
+  where flow_slug = p_flow_slug;
+
+  -- Set timestamps for runs
+  update pgflow.runs
+  set
+    started_at = now() - interval '2 days' - (days_old * interval '1 day'),
+    completed_at = now() - (days_old * interval '1 day'),
+    status = 'completed',
+    remaining_steps = 0
+  where flow_slug = p_flow_slug;
+end;
+$$;
+
+-- Set timestamps for failed runs/steps/tasks to be older than the cutoff
+create or replace function pgflow_tests.set_failed_flow_timestamps(
+  p_flow_slug text,
+  days_old integer
+) returns void language plpgsql as $$
+begin
+  -- Set timestamps for step_tasks
+  update pgflow.step_tasks
+  set
+    queued_at = now() - interval '1 day' - (days_old * interval '1 day'),
+    failed_at = now() - (days_old * interval '1 day'),
+    status = 'failed',
+    error_message = 'Test failure'
+  where flow_slug = p_flow_slug;
+
+  -- Set timestamps for step_states
+  update pgflow.step_states
+  set
+    created_at = now() - interval '2 days' - (days_old * interval '1 day'),
+    started_at = now() - interval '1 day' - (days_old * interval '1 day'),
+    failed_at = now() - (days_old * interval '1 day'),
+    status = 'failed'
+  where flow_slug = p_flow_slug;
+
+  -- Set timestamps for runs
+  update pgflow.runs
+  set
+    started_at = now() - interval '2 days' - (days_old * interval '1 day'),
+    failed_at = now() - (days_old * interval '1 day'),
+    status = 'failed'
+  where flow_slug = p_flow_slug;
+end;
+$$;
+
+-- Set timestamps for running flows to be older than the cutoff
+create or replace function pgflow_tests.set_running_flow_timestamps(
+  p_flow_slug text,
+  days_old integer
+) returns void language plpgsql as $$
+begin
+  -- Set timestamps for step_tasks
+  update pgflow.step_tasks
+  set
+    queued_at = now() - (days_old * interval '1 day'),
+    status = 'queued'
+  where flow_slug = p_flow_slug;
+
+  -- Set timestamps for step_states
+  update pgflow.step_states
+  set
+    created_at = now() - (days_old * interval '1 day'),
+    started_at = now() - (days_old * interval '1 day'),
+    status = 'started'
+  where flow_slug = p_flow_slug;
+
+  -- Set timestamps for runs
+  update pgflow.runs
+  set
+    started_at = now() - (days_old * interval '1 day'),
+    status = 'started'
+  where flow_slug = p_flow_slug;
+end;
+$$;

pkgs/core/supabase/tests/_shared/prune_data_older_than.sql.raw

@@ -0,0 +1,62 @@
+/**
+ * Prunes old records from pgflow tables and PGMQ archive tables.
+ *
+ * @param retention_interval - Interval of recent records to keep (e.g., interval '28 days', interval '3 months')
+ */
+create or replace function pgflow.prune_data_older_than(
+  retention_interval INTERVAL
+) returns void language plpgsql as $$
+DECLARE
+  cutoff_timestamp TIMESTAMPTZ := now() - retention_interval;
+  flow_record RECORD;
+  archive_table TEXT;
+  dynamic_sql TEXT;
+BEGIN
+  -- Delete old worker records
+  DELETE FROM pgflow.workers
+  WHERE last_heartbeat_at < cutoff_timestamp;
+
+  -- Delete old step_tasks records
+  DELETE FROM pgflow.step_tasks
+  WHERE (
+    (completed_at IS NOT NULL AND completed_at < cutoff_timestamp) OR
+    (failed_at IS NOT NULL AND failed_at < cutoff_timestamp)
+  );
+
+  -- Delete old step_states records
+  DELETE FROM pgflow.step_states
+  WHERE (
+    (completed_at IS NOT NULL AND completed_at < cutoff_timestamp) OR
+    (failed_at IS NOT NULL AND failed_at < cutoff_timestamp)
+  );
+
+  -- Delete old runs records
+  DELETE FROM pgflow.runs
+  WHERE (
+    (completed_at IS NOT NULL AND completed_at < cutoff_timestamp) OR
+    (failed_at IS NOT NULL AND failed_at < cutoff_timestamp)
+  );
+
+  -- Prune archived messages from PGMQ archive tables (pgmq.a_{flow_slug})
+  -- For each flow, delete old archived messages
+  FOR flow_record IN SELECT DISTINCT flow_slug FROM pgflow.flows
+  LOOP
+    -- Build the archive table name
+    archive_table := pgmq.format_table_name(flow_record.flow_slug, 'a');
+
+    -- Check if the archive table exists
+    IF EXISTS (
+      SELECT 1 FROM information_schema.tables
+      WHERE table_schema = 'pgmq' AND table_name = archive_table
+    ) THEN
+      -- Build and execute a dynamic SQL statement to delete old archive records
+      dynamic_sql := format('
+        DELETE FROM pgmq.%I
+        WHERE archived_at < $1
+      ', archive_table);
+
+      EXECUTE dynamic_sql USING cutoff_timestamp;
+    END IF;
+  END LOOP;
+END
+$$;

pkgs/core/supabase/tests/maintenance/prune_archived_messages.test.sql

@@ -0,0 +1,70 @@
+begin;
+select plan(4);
+select pgflow_tests.reset_db();
+
+-- Load the prune_data_older_than function
+\i _shared/prune_data_older_than.sql.raw
+
+-- Create test flows with sequential structure to ensure message archiving
+select pgflow_tests.setup_flow('sequential');
+
+-- Start the flow and complete the first step to create an archived message
+select pgflow.start_flow('sequential', '{"test": true}'::JSONB);
+
+-- Get the run_id for the first flow
+\set run_id_1 `echo "SELECT run_id FROM pgflow.runs LIMIT 1;" | psql -t -A`
+
+-- Complete the first task using poll_and_complete to ensure proper handling
+select pgflow_tests.poll_and_complete('sequential');
+
+-- Verify that a message was archived
+select is(
+  (select count(*)::INT from pgmq.a_sequential),
+  1::INT,
+  'One message should be archived initially'
+);
+
+-- Create a second flow and complete its first task to archive another message
+select pgflow.start_flow('sequential', '{"test": "second run"}'::JSONB);
+select pgflow_tests.poll_and_complete('sequential');
+
+-- Now we have two archived messages
+select is(
+  (select count(*)::INT from pgmq.a_sequential),
+  2::INT,
+  'Two messages should be archived'
+);
+
+-- Set different timestamps for the archived messages
+-- Make one message old (31 days)
+update pgmq.a_sequential
+set archived_at = now() - INTERVAL '31 days'
+where msg_id = (select min(msg_id) from pgmq.a_sequential);
+
+-- Leave the other message recent (5 days)
+update pgmq.a_sequential
+set archived_at = now() - INTERVAL '5 days'
+where msg_id = (select max(msg_id) from pgmq.a_sequential);
+
+-- Prune with 30-day retention
+select pgflow.prune_data_older_than(make_interval(days => 30));
+
+-- TEST: Only the old archived message should be pruned
+select is(
+  (select count(*)::INT from pgmq.a_sequential),
+  1::INT,
+  'Only one message should remain after pruning'
+);
+
+-- TEST: Check that the remaining message is the recent one
+select is(
+  (
+    select (extract(day from now() - archived_at) < 10)::BOOLEAN
+    from pgmq.a_sequential
+  ),
+  true,
+  'The remaining message should be the recent one'
+);
+
+select finish();
+rollback;

pkgs/core/supabase/tests/maintenance/prune_completed_flows.test.sql

@@ -0,0 +1,69 @@
+begin;
+select plan(4);
+select pgflow_tests.reset_db();
+
+-- Load the prune_data_older_than function
+\i _shared/prune_data_older_than.sql.raw
+
+-- Create test flows
+select pgflow.create_flow('old_completed_flow', max_attempts => 0);
+select pgflow.add_step('old_completed_flow', 'step');
+
+select pgflow.create_flow('recent_completed_flow', max_attempts => 0);
+select pgflow.add_step('recent_completed_flow', 'step');
+
+select pgflow.create_flow('running_old_flow', max_attempts => 0);
+select pgflow.add_step('running_old_flow', 'step');
+
+-- Start and complete flows
+select pgflow.start_flow('old_completed_flow', '{}'::jsonb);
+select pgflow_tests.poll_and_complete('old_completed_flow');
+
+select pgflow.start_flow('recent_completed_flow', '{}'::jsonb);
+select pgflow_tests.poll_and_complete('recent_completed_flow');
+
+select pgflow.start_flow('running_old_flow', '{}'::jsonb);
+
+-- Set timestamps to simulate age
+-- Old completed flow: 31 days old
+select pgflow_tests.set_completed_flow_timestamps('old_completed_flow', 31);
+
+-- Recent completed flow: 25 days old (within 30-day retention)
+select pgflow_tests.set_completed_flow_timestamps('recent_completed_flow', 25);
+
+-- Running flow: 40 days old but still running (should not be pruned)
+select pgflow_tests.set_running_flow_timestamps('running_old_flow', 40);
+
+-- Verify setup
+select is(
+  (select count(*) from pgflow.runs),
+  3::bigint,
+  'Should have 3 runs before pruning'
+);
+
+-- Prune old records with 30-day retention
+select pgflow.prune_data_older_than(make_interval(days => 30));
+
+-- TEST: Only the old completed flow should be pruned
+select is(
+  (select array_agg(flow_slug order by flow_slug) from pgflow.runs),
+  array['recent_completed_flow', 'running_old_flow'],
+  'Only old completed flow should be pruned'
+);
+
+-- TEST: Step states for old completed flow should be pruned
+select is(
+  (select count(*) from pgflow.step_states where flow_slug = 'old_completed_flow'),
+  0::bigint,
+  'Step states for old completed flow should be pruned'
+);
+
+-- TEST: Step tasks for old completed flow should be pruned
+select is(
+  (select count(*) from pgflow.step_tasks where flow_slug = 'old_completed_flow'),
+  0::bigint,
+  'Step tasks for old completed flow should be pruned'
+);
+
+select finish();
+rollback;

pkgs/core/supabase/tests/maintenance/prune_failed_flows.test.sql

@@ -0,0 +1,69 @@
+begin;
+select plan(4);
+select pgflow_tests.reset_db();
+
+-- Load the prune_data_older_than function
+\i _shared/prune_data_older_than.sql.raw
+
+-- Create test flows
+select pgflow.create_flow('old_failed_flow', max_attempts => 0);
+select pgflow.add_step('old_failed_flow', 'step');
+
+select pgflow.create_flow('recent_failed_flow', max_attempts => 0);
+select pgflow.add_step('recent_failed_flow', 'step');
+
+select pgflow.create_flow('running_old_flow', max_attempts => 0);
+select pgflow.add_step('running_old_flow', 'step');
+
+-- Start and fail flows
+select pgflow.start_flow('old_failed_flow', '{}'::jsonb);
+select pgflow_tests.poll_and_fail('old_failed_flow');
+
+select pgflow.start_flow('recent_failed_flow', '{}'::jsonb);
+select pgflow_tests.poll_and_fail('recent_failed_flow');
+
+select pgflow.start_flow('running_old_flow', '{}'::jsonb);
+
+-- Set timestamps to simulate age
+-- Old failed flow: 31 days old
+select pgflow_tests.set_failed_flow_timestamps('old_failed_flow', 31);
+
+-- Recent failed flow: 25 days old (within 30-day retention)
+select pgflow_tests.set_failed_flow_timestamps('recent_failed_flow', 25);
+
+-- Running flow: 40 days old but still running (should not be pruned)
+select pgflow_tests.set_running_flow_timestamps('running_old_flow', 40);
+
+-- Verify setup
+select is(
+  (select count(*) from pgflow.runs),
+  3::bigint,
+  'Should have 3 runs before pruning'
+);
+
+-- Prune old records with 30-day retention
+select pgflow.prune_data_older_than(make_interval(days => 30));
+
+-- TEST: Only the old failed flow should be pruned
+select is(
+  (select array_agg(flow_slug order by flow_slug) from pgflow.runs),
+  array['recent_failed_flow', 'running_old_flow'],
+  'Only old failed flow should be pruned'
+);
+
+-- TEST: Step states for old failed flow should be pruned
+select is(
+  (select count(*) from pgflow.step_states where flow_slug = 'old_failed_flow'),
+  0::bigint,
+  'Step states for old failed flow should be pruned'
+);
+
+-- TEST: Step tasks for old failed flow should be pruned
+select is(
+  (select count(*) from pgflow.step_tasks where flow_slug = 'old_failed_flow'),
+  0::bigint,
+  'Step tasks for old failed flow should be pruned'
+);
+
+select finish();
+rollback;