feat: add allowDataLoss parameter to flow compilation (#545)

# Add `allow_data_loss` Parameter to Flow Compilation

This PR adds a new `allow_data_loss` parameter to the `pgflow.ensure_flow_compiled` function, enabling destructive flow recompilation in production environments when explicitly enabled.

## Key Changes

- Added an optional `allow_data_loss` boolean parameter (default: false) to the `pgflow.ensure_flow_compiled` function
- Modified the recompilation logic to check `v_is_local OR allow_data_loss` before allowing destructive operations
- Updated the edge worker API to support a new `compilation` configuration object that can include `allowDataLoss: true`
- Replaced the deprecated `ensureCompiledOnStartup` boolean with the more flexible `compilation` config
- Added comprehensive tests for the new parameter in both SQL and TypeScript

## Benefits

- Provides a safe way to iterate on flow designs in production environments when needed
- Maintains the default safe behavior (no data loss in production) while adding an explicit opt-in for destructive changes
- Improves developer experience by allowing controlled recompilation without environment switching

## Usage

```typescript
// Allow destructive recompilation in any environment
const worker = createFlowWorker(MyFlow, {
  compilation: { allowDataLoss: true },
  // other config...
});

// Skip compilation entirely (pre-compiled flows)
const worker = createFlowWorker(MyFlow, {
  compilation: false,
  // other config...
});
```

Author: jumski

pkgs/core/schemas/0100_function_ensure_flow_compiled.sql

@@ -3,7 +3,8 @@
 -- Returns: { status: 'compiled' | 'verified' | 'recompiled' | 'mismatch', differences: text[] }
 create or replace function pgflow.ensure_flow_compiled(
   flow_slug text,
-  shape jsonb
+  shape jsonb,
+  allow_data_loss boolean default false
 )
 returns jsonb
 language plpgsql
@@ -48,7 +49,7 @@ BEGIN
   -- 6. Shapes differ - auto-detect environment via is_local()
   v_is_local := pgflow.is_local();
 
-  IF v_is_local THEN
+  IF v_is_local OR allow_data_loss THEN
     -- Recompile in local/dev: full deletion + fresh compile
     PERFORM pgflow.delete_flow_and_data(ensure_flow_compiled.flow_slug);
     PERFORM pgflow._create_flow_from_shape(ensure_flow_compiled.flow_slug, ensure_flow_compiled.shape);

pkgs/core/src/database-types.ts

@@ -485,7 +485,7 @@ export type Database = {
         Returns: undefined
       }
       ensure_flow_compiled: {
-        Args: { flow_slug: string; shape: Json }
+        Args: { allow_data_loss?: boolean; flow_slug: string; shape: Json }
         Returns: Json
       }
       ensure_workers: {

pkgs/core/supabase/migrations/20251212100113_pgflow_allow_data_loss_parameter.sql

@@ -0,0 +1,54 @@
+-- Drop old 2-parameter version before creating new 3-parameter version
+DROP FUNCTION IF EXISTS "pgflow"."ensure_flow_compiled" (text, jsonb);
+
+-- Create "ensure_flow_compiled" function with allow_data_loss parameter
+CREATE FUNCTION "pgflow"."ensure_flow_compiled" ("flow_slug" text, "shape" jsonb, "allow_data_loss" boolean DEFAULT false) RETURNS jsonb LANGUAGE plpgsql SET "search_path" = '' AS $$
+DECLARE
+  v_lock_key int;
+  v_flow_exists boolean;
+  v_db_shape jsonb;
+  v_differences text[];
+  v_is_local boolean;
+BEGIN
+  -- Generate lock key from flow_slug (deterministic hash)
+  v_lock_key := hashtext(ensure_flow_compiled.flow_slug);
+
+  -- Acquire transaction-level advisory lock
+  -- Serializes concurrent compilation attempts for same flow
+  PERFORM pg_advisory_xact_lock(1, v_lock_key);
+
+  -- 1. Check if flow exists
+  SELECT EXISTS(SELECT 1 FROM pgflow.flows AS flow WHERE flow.flow_slug = ensure_flow_compiled.flow_slug)
+  INTO v_flow_exists;
+
+  -- 2. If flow missing: compile (both environments)
+  IF NOT v_flow_exists THEN
+    PERFORM pgflow._create_flow_from_shape(ensure_flow_compiled.flow_slug, ensure_flow_compiled.shape);
+    RETURN jsonb_build_object('status', 'compiled', 'differences', '[]'::jsonb);
+  END IF;
+
+  -- 3. Get current shape from DB
+  v_db_shape := pgflow._get_flow_shape(ensure_flow_compiled.flow_slug);
+
+  -- 4. Compare shapes
+  v_differences := pgflow._compare_flow_shapes(ensure_flow_compiled.shape, v_db_shape);
+
+  -- 5. If shapes match: return verified
+  IF array_length(v_differences, 1) IS NULL THEN
+    RETURN jsonb_build_object('status', 'verified', 'differences', '[]'::jsonb);
+  END IF;
+
+  -- 6. Shapes differ - auto-detect environment via is_local()
+  v_is_local := pgflow.is_local();
+
+  IF v_is_local OR allow_data_loss THEN
+    -- Recompile in local/dev: full deletion + fresh compile
+    PERFORM pgflow.delete_flow_and_data(ensure_flow_compiled.flow_slug);
+    PERFORM pgflow._create_flow_from_shape(ensure_flow_compiled.flow_slug, ensure_flow_compiled.shape);
+    RETURN jsonb_build_object('status', 'recompiled', 'differences', to_jsonb(v_differences));
+  ELSE
+    -- Fail in production
+    RETURN jsonb_build_object('status', 'mismatch', 'differences', to_jsonb(v_differences));
+  END IF;
+END;
+$$;

pkgs/core/supabase/migrations/atlas.sum

@@ -1,4 +1,4 @@
-h1:s6n1PLk8xdvwsEbFcEbLUMaRj0zxGmAfBcaVkk3GKak=
+h1:yUEa4QnI0kEuEtxWNiNgRgf9GENvG1FkUmzwlu0vLm0=
 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=
@@ -13,3 +13,4 @@ h1:s6n1PLk8xdvwsEbFcEbLUMaRj0zxGmAfBcaVkk3GKak=
 20251104080523_pgflow_upgrade_pgmq_1_5_1.sql h1:Fw7zpMWnjhAHQ0qBJAprAvGl7dJMd8ExNHg8aKvkzTg=
 20251130000000_pgflow_auto_compilation.sql h1:qs+3qq1Vsyo0ETzbxDnmkVtSUa6XHkd/K9wF/3W46jM=
 20251209074533_pgflow_worker_management.sql h1:ozFkYM1EvEH7dGvW+1pWgpzbwdWlwuQoddqLomL1GXw=
+20251212100113_pgflow_allow_data_loss_parameter.sql h1:Fg3RHj51STNHS4epQ2J4AFMj7NwG0XfyDTSA/9dcBIQ=

pkgs/core/supabase/tests/ensure_flow_compiled/allow_data_loss_default_false.test.sql

@@ -0,0 +1,55 @@
+begin;
+select plan(3);
+select pgflow_tests.reset_db();
+
+-- Setup: Simulate production environment (not local)
+select set_config('app.settings.jwt_secret', 'production-jwt-secret-that-differs-from-local', true);
+
+-- Setup: Create flow with different shape
+select pgflow.create_flow('default_loss_flow');
+select pgflow.add_step('default_loss_flow', 'old_step');
+
+-- Test: Different shape should return mismatch when allow_data_loss defaults to false
+select is(
+  (
+    select result->>'status'
+    from pgflow.ensure_flow_compiled(
+      'default_loss_flow',
+      '{
+        "steps": [
+          {"slug": "new_step", "stepType": "single", "dependencies": []}
+        ]
+      }'::jsonb,
+      false  -- allow_data_loss = false (explicit)
+    ) as result
+  ),
+  'mismatch',
+  'Should return mismatch when allow_data_loss=false in production'
+);
+
+-- Verify differences are returned
+select ok(
+  (
+    select jsonb_array_length(result->'differences') > 0
+    from pgflow.ensure_flow_compiled(
+      'default_loss_flow',
+      '{
+        "steps": [
+          {"slug": "new_step", "stepType": "single", "dependencies": []}
+        ]
+      }'::jsonb,
+      false
+    ) as result
+  ),
+  'Should return differences when allow_data_loss=false'
+);
+
+-- Verify database was NOT modified
+select is(
+  (select step_slug from pgflow.steps where flow_slug = 'default_loss_flow'),
+  'old_step',
+  'Database should not be modified when allow_data_loss=false'
+);
+
+select finish();
+rollback;

pkgs/core/supabase/tests/ensure_flow_compiled/allow_data_loss_recompiles.test.sql

@@ -0,0 +1,45 @@
+begin;
+select plan(3);
+select pgflow_tests.reset_db();
+
+-- Setup: Simulate production environment (not local)
+select set_config('app.settings.jwt_secret', 'production-jwt-secret-that-differs-from-local', true);
+
+-- Setup: Create flow with different shape
+select pgflow.create_flow('allow_loss_flow');
+select pgflow.add_step('allow_loss_flow', 'old_step');
+
+-- Test: Different shape should recompile when allow_data_loss=true even when is_local()=false
+select is(
+  (
+    select result->>'status'
+    from pgflow.ensure_flow_compiled(
+      'allow_loss_flow',
+      '{
+        "steps": [
+          {"slug": "new_step", "stepType": "single", "dependencies": []}
+        ]
+      }'::jsonb,
+      true  -- allow_data_loss = true
+    ) as result
+  ),
+  'recompiled',
+  'Should recompile when allow_data_loss=true even in production'
+);
+
+-- Verify old step is gone
+select is(
+  (select count(*)::int from pgflow.steps where flow_slug = 'allow_loss_flow' and step_slug = 'old_step'),
+  0,
+  'Old step should be deleted when allow_data_loss=true'
+);
+
+-- Verify new step exists
+select is(
+  (select count(*)::int from pgflow.steps where flow_slug = 'allow_loss_flow' and step_slug = 'new_step'),
+  1,
+  'New step should be created when allow_data_loss=true'
+);
+
+select finish();
+rollback;

pkgs/edge-worker/src/core/Queries.ts

@@ -54,7 +54,8 @@ export class Queries {
 
   async ensureFlowCompiled(
     flowSlug: string,
-    shape: FlowShape
+    shape: FlowShape,
+    allowDataLoss = false
   ): Promise<EnsureFlowCompiledResult> {
     // SAFETY: FlowShape is JSON-compatible by construction (only strings, numbers,
     // arrays, and plain objects), but TypeScript can't prove this because FlowShape
@@ -68,7 +69,8 @@ export class Queries {
     const [result] = await this.sql<{ result: EnsureFlowCompiledResult }[]>`
       SELECT pgflow.ensure_flow_compiled(
         ${flowSlug},
-        ${shapeJson}::jsonb
+        ${shapeJson}::jsonb,
+        ${allowDataLoss}
       ) as result
     `;
     return result.result;

pkgs/edge-worker/src/core/workerConfigTypes.ts

@@ -140,17 +140,31 @@ export type QueueWorkerConfig = {
  */
 export type ResolvedQueueWorkerConfig = Required<Omit<QueueWorkerConfig, 'retryDelay' | 'retryLimit'>>;
 
+/**
+ * Compilation configuration for flow workers.
+ */
+export interface CompilationConfig {
+  /**
+   * Allow destructive recompilation on shape mismatch.
+   * When true, ALL flow data (runs, tasks, history) will be deleted
+   * if the flow shape changes. Use only during iteration phases (e.g., Lovable).
+   * When false (default), recompilation only happens in local environment.
+   * @default false
+   */
+  allowDataLoss?: boolean;
+}
+
 /**
  * Configuration for the flow worker with two-phase polling
  */
 export type FlowWorkerConfig = {
   /**
-   * Whether to verify/compile flow at worker startup.
-   * When true (default), worker calls pgflow.ensure_flow_compiled() before polling.
-   * Set to false to skip compilation check (useful if flows are pre-compiled via CLI).
-   * @default true
+   * Compilation behavior at worker startup.
+   * - undefined or {}: Auto-detect (local=recompile, prod=fail on mismatch)
+   * - false: Skip compilation check (flows pre-compiled via CLI)
+   * - { allowDataLoss: true }: Allow destructive recompile anywhere
    */
-  ensureCompiledOnStartup?: boolean;
+  compilation?: false | CompilationConfig;
 
   /**
    * How many tasks are processed at the same time
@@ -209,7 +223,7 @@ export type FlowWorkerConfig = {
 /**
  * Resolved flow configuration with all defaults applied
  */
-export type ResolvedFlowWorkerConfig = Required<Omit<FlowWorkerConfig, 'connectionString' | 'env' | 'ensureCompiledOnStartup'>> & {
+export type ResolvedFlowWorkerConfig = Required<Omit<FlowWorkerConfig, 'connectionString' | 'env' | 'compilation'>> & {
   connectionString: string | undefined;
   env: Record<string, string | undefined>;
 };

pkgs/edge-worker/src/flow/FlowWorkerLifecycle.ts

@@ -6,9 +6,11 @@ import type { AnyFlow } from '@pgflow/dsl';
 import { extractFlowShape } from '@pgflow/dsl';
 import { FlowShapeMismatchError } from './errors.js';
 
+import type { CompilationConfig } from '../core/workerConfigTypes.js';
+
 export interface FlowLifecycleConfig {
   heartbeatInterval?: number;
-  ensureCompiledOnStartup?: boolean;
+  compilation?: false | CompilationConfig;
 }
 
 /**
@@ -30,15 +32,16 @@ export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements ILifecycle {
   private _edgeFunctionName?: string;
   private heartbeatInterval: number;
   private lastHeartbeat = 0;
-  private ensureCompiledOnStartup: boolean;
+  private compilation: false | CompilationConfig;
 
   constructor(queries: Queries, flow: TFlow, logger: Logger, config?: FlowLifecycleConfig) {
     this.queries = queries;
     this.flow = flow;
     this.logger = logger;
     this.workerState = new WorkerState(logger);
     this.heartbeatInterval = config?.heartbeatInterval ?? 5000;
-    this.ensureCompiledOnStartup = config?.ensureCompiledOnStartup ?? true;
+    // Default to {} (enable compilation with default settings) if not specified
+    this.compilation = config?.compilation ?? {};
   }
 
   async acknowledgeStart(workerBootstrap: WorkerBootstrap): Promise<void> {
@@ -53,10 +56,11 @@ export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements ILifecycle {
 
     // Compile/verify flow as part of Starting (before registering worker)
     let compilationStatus: CompilationStatus = 'verified';
-    if (this.ensureCompiledOnStartup) {
-      compilationStatus = await this.ensureFlowCompiled();
+    if (this.compilation !== false) {
+      const allowDataLoss = this.compilation.allowDataLoss ?? false;
+      compilationStatus = await this.ensureFlowCompiled(allowDataLoss);
     } else {
-      this.logger.info(`Skipping compilation check for flow '${this.flow.slug}' (ensureCompiledOnStartup=false)`);
+      this.logger.info(`Skipping compilation check for flow '${this.flow.slug}' (compilation: false)`);
     }
 
     // Log startup banner with compilation status
@@ -71,12 +75,13 @@ export class FlowWorkerLifecycle<TFlow extends AnyFlow> implements ILifecycle {
     this.workerState.transitionTo(States.Running);
   }
 
-  private async ensureFlowCompiled(): Promise<CompilationStatus> {
+  private async ensureFlowCompiled(allowDataLoss: boolean): Promise<CompilationStatus> {
     const shape = extractFlowShape(this.flow);
 
     const result = await this.queries.ensureFlowCompiled(
       this.flow.slug,
-      shape
+      shape,
+      allowDataLoss
     );
 
     if (result.status === 'mismatch') {

pkgs/edge-worker/src/flow/createFlowWorker.ts

@@ -91,7 +91,7 @@ export function createFlowWorker<TFlow extends AnyFlow, TResources extends Recor
     flow,
     createLogger('FlowWorkerLifecycle'),
     {
-      ensureCompiledOnStartup: config.ensureCompiledOnStartup ?? true
+      compilation: config.compilation
     }
   );