feat(edge-worker): add workerConfig to handler execution context (#203)

Adds workerConfig to handler execution context, allowing handlers to access worker configuration for intelligent decision-making based on retry limits, concurrency settings, and other parameters.
Key Changes

    Added workerConfig to MessageExecution and StepTaskExecution contexts
    Deep frozen config prevents accidental handler modifications while keeping worker config mutable
    Performance optimized - config frozen once at startup and cached for reuse
    Reorganized configuration handling with cleaner factory methods and consolidated defaults
    Simplified EdgeWorker API by removing unnecessary union types
    Environment variables always sourced from platform (users cannot override)

Author: jumski

.changeset/bright-clocks-bet.md

@@ -0,0 +1,14 @@
+---
+'@pgflow/edge-worker': patch
+---
+
+Add workerConfig to handler execution context
+
+Handlers can now access worker configuration through `context.workerConfig` to make intelligent decisions based on retry limits, concurrency settings, and other worker parameters. The config is deeply frozen to prevent accidental modifications while remaining mutable by the worker itself for future features.
+
+Key improvements:
+- Added `workerConfig` to MessageExecution and StepTaskExecution contexts
+- Config is cached and frozen once at startup for optimal performance
+- Reorganized configuration handling with cleaner factory methods
+- Simplified EdgeWorker API by removing unnecessary union types
+- Environment variables always sourced from platform (users cannot override)

pkgs/edge-worker/src/EdgeWorker.ts

@@ -13,12 +13,6 @@ import type { MessageHandlerFn } from './queue/types.js';
 import type { AnyFlow } from '@pgflow/dsl';
 import type { CompatibleFlow } from './types/flowCompatibility.js';
 
-/**
- * Configuration options for the EdgeWorker.
- */
-export type EdgeWorkerConfig =
-  | Omit<QueueWorkerConfig, 'sql'>
-  | Omit<FlowWorkerConfig, 'sql'>;
 
 /**
  * EdgeWorker is the main entry point for creating and starting edge workers.
@@ -88,17 +82,17 @@ export class EdgeWorker {
     TFlow extends AnyFlow = AnyFlow
   >(
     handlerOrFlow: MessageHandlerFn<TPayload> | TFlow,
-    config: EdgeWorkerConfig = {}
-  ): Promise<void> {
+    config?: Omit<QueueWorkerConfig, 'sql'> | Omit<FlowWorkerConfig, 'sql'>
+  ): Promise<PlatformAdapter> {
     if (typeof handlerOrFlow === 'function') {
-      await this.startQueueWorker(
+      return await this.startQueueWorker(
         handlerOrFlow as MessageHandlerFn<TPayload>,
-        config as QueueWorkerConfig
+        config
       );
     } else {
-      await this.startFlowWorker(
+      return await this.startFlowWorker(
         handlerOrFlow as TFlow,
-        config as FlowWorkerConfig
+        config
       );
     }
   }
@@ -142,21 +136,15 @@ export class EdgeWorker {
   static async startQueueWorker<TPayload extends Json = Json>(
     handler: MessageHandlerFn<TPayload>,
     config: QueueWorkerConfig = {}
-  ) {
+  ): Promise<PlatformAdapter> {
     this.ensureFirstCall();
 
     // First, create the adapter
     this.platform = await createAdapter();
 
-    // Apply default values to the config
+    // Add platform-specific values to the config
     const workerConfig: QueueWorkerConfig = {
       ...config,
-      queueName: config.queueName || 'tasks',
-      maxConcurrent: config.maxConcurrent ?? 10,
-      maxPgConnections: config.maxPgConnections ?? 4,
-      maxPollSeconds: config.maxPollSeconds ?? 5,
-      pollIntervalMs: config.pollIntervalMs ?? 200,
-      visibilityTimeout: config.visibilityTimeout ?? 10,
       connectionString:
         config.connectionString || this.platform.connectionString,
       env: this.platform.env,
@@ -165,6 +153,8 @@ export class EdgeWorker {
     await this.platform.startWorker((createLoggerFn) => {
       return createQueueWorker(handler, workerConfig, createLoggerFn, this.platform);
     });
+
+    return this.platform;
   }
 
   /**
@@ -196,27 +186,24 @@ export class EdgeWorker {
   static async startFlowWorker<TFlow extends AnyFlow>(
     flow: CompatibleFlow<TFlow>,
     config: FlowWorkerConfig = {}
-  ) {
+  ): Promise<PlatformAdapter> {
     this.ensureFirstCall();
 
     // First, create the adapter
     this.platform = await createAdapter();
 
-    // Apply default values to the config
+    // Add platform-specific values to the config
     const workerConfig: FlowWorkerConfig = {
       ...config,
-      maxConcurrent: config.maxConcurrent ?? 10,
-      maxPgConnections: config.maxPgConnections ?? 4,
-      batchSize: config.batchSize ?? 10,
-      maxPollSeconds: config.maxPollSeconds ?? 2,
-      pollIntervalMs: config.pollIntervalMs ?? 100,
       connectionString:
         config.connectionString || this.platform.connectionString,
     };
 
     await this.platform.startWorker((createLoggerFn) => {
       return createFlowWorker(flow, workerConfig, createLoggerFn, this.platform);
     });
+
+    return this.platform;
   }
 
   /**

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

@@ -3,6 +3,7 @@ import type { BaseContext, AnyFlow, AllStepInputs } from '@pgflow/dsl';
 import type { Json } from './types.js';
 import type { PgmqMessageRecord } from '../queue/types.js';
 import type { StepTaskRecord } from '../flow/types.js';
+import type { QueueWorkerConfig, FlowWorkerConfig } from './workerConfigTypes.js';
 
 /* ──────────────────────────────────────────────────────────────────────
    1.  PLATFORM LAYER
@@ -22,11 +23,13 @@ export type PlatformContext<
 
 export interface MessageExecution<TPayload extends Json = Json> {
   rawMessage: PgmqMessageRecord<TPayload>;
+  workerConfig: Readonly<Omit<QueueWorkerConfig, 'sql'>>;
 }
 
 export interface StepTaskExecution<TFlow extends AnyFlow = AnyFlow> {
   rawMessage: PgmqMessageRecord<AllStepInputs<TFlow>>;
   stepTask  : StepTaskRecord<TFlow>;
+  workerConfig: Readonly<Omit<FlowWorkerConfig, 'sql'>>;
 }
 
 /** Message handler context for any platform */
@@ -51,6 +54,20 @@ export interface StepTaskWithMessage<TFlow extends AnyFlow> {
   task   : StepTaskRecord<TFlow>;
 }
 
+import { deepClone, deepFreeze } from './deepUtils.js';
+
+/**
+ * Creates a context-safe version of worker config by excluding sql connection,
+ * deep cloning, and deep freezing the result to prevent modification by handlers
+ */
+export function createContextSafeConfig<T extends Record<string, unknown>>(
+  config: T
+): Readonly<T extends { sql: unknown } ? Omit<T, 'sql'> : T> {
+  const { sql: _sql, ...safeConfig } = config as T & { sql?: unknown };
+  const clonedConfig = deepClone(safeConfig);
+  return deepFreeze(clonedConfig) as Readonly<T extends { sql: unknown } ? Omit<T, 'sql'> : T>;
+}
+
 /* ──────────────────────────────────────────────────────────────────────
    4.  LEGACY COMPATIBILITY (for backward compatibility only)
    --------------------------------------------------------------------- */

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

@@ -0,0 +1,40 @@
+/**
+ * Deep clone an object recursively
+ */
+export function deepClone<T>(obj: T): T {
+  if (obj === null || typeof obj !== 'object') {
+    return obj;
+  }
+
+  if (obj instanceof Date) {
+    return new Date(obj.getTime()) as T;
+  }
+
+  const source = obj as Record<PropertyKey, unknown>;
+  const target = Array.isArray(obj) ? [] as unknown[] : {} as Record<PropertyKey, unknown>;
+  
+  for (const key of Reflect.ownKeys(source)) {
+    (target as Record<PropertyKey, unknown>)[key] = deepClone(source[key]);
+  }
+  
+  return target as T;
+}
+
+/**
+ * Deep freeze an object recursively to prevent any modifications
+ */
+export function deepFreeze<T>(obj: T): T {
+  // Get property names
+  const propNames = Reflect.ownKeys(obj as object);
+
+  // Freeze properties before freezing self
+  for (const name of propNames) {
+    const value = (obj as Record<PropertyKey, unknown>)[name];
+
+    if ((value && typeof value === 'object') || typeof value === 'function') {
+      deepFreeze(value);
+    }
+  }
+
+  return Object.freeze(obj);
+}

pkgs/edge-worker/src/core/supabase-test-utils.ts

@@ -6,6 +6,8 @@ import type { PgmqMessageRecord } from '../queue/types.js';
 import type { 
   StepTaskWithMessage
 } from './context.js';
+import { createContextSafeConfig } from './context.js';
+import type { QueueWorkerConfig, FlowWorkerConfig } from './workerConfigTypes.js';
 import type { SupabaseEnv } from '@pgflow/dsl/supabase';
 import { createServiceSupabaseClient } from './supabase-utils.js';
 
@@ -18,8 +20,9 @@ export function createQueueWorkerContext<TPayload extends Json>(params: {
   sql: Sql;
   abortSignal: AbortSignal;
   rawMessage: PgmqMessageRecord<TPayload>;
+  workerConfig?: Readonly<Omit<QueueWorkerConfig, 'sql'>>;
 }) {
-  const { env, sql, abortSignal, rawMessage } = params;
+  const { env, sql, abortSignal, rawMessage, workerConfig } = params;
 
   // Create Supabase client if env vars exist, otherwise create mock
   const supabase = env.SUPABASE_URL && env.SUPABASE_SERVICE_ROLE_KEY
@@ -36,6 +39,7 @@ export function createQueueWorkerContext<TPayload extends Json>(params: {
 
     // Message execution context
     rawMessage,
+    ...(workerConfig && { workerConfig: createContextSafeConfig(workerConfig) }),
 
     // Supabase-specific resources (always present in Phase 1)
     sql,
@@ -53,8 +57,9 @@ export function createFlowWorkerContext<TFlow extends AnyFlow = AnyFlow>(params:
   sql: Sql;
   abortSignal: AbortSignal;
   taskWithMessage?: StepTaskWithMessage<TFlow>;
+  workerConfig?: Readonly<Omit<FlowWorkerConfig, 'sql'>>;
 }) {
-  const { env, sql, abortSignal, taskWithMessage } = params;
+  const { env, sql, abortSignal, taskWithMessage, workerConfig } = params;
 
   // Create Supabase client if env vars exist, otherwise create mock
   const supabase = env.SUPABASE_URL && env.SUPABASE_SERVICE_ROLE_KEY
@@ -76,6 +81,7 @@ export function createFlowWorkerContext<TFlow extends AnyFlow = AnyFlow>(params:
     // Step task execution context
     rawMessage: taskWithMessage.message,
     stepTask: taskWithMessage.task,
+    ...(workerConfig && { workerConfig: createContextSafeConfig(workerConfig) }),
 
     // Supabase-specific resources (always present in Phase 1)
     sql,