pgflow-dev
diff --git a/‎.changeset/clean-zoos-smoke.md
Lines changed: 9 additions & 0 deletions b/‎.changeset/clean-zoos-smoke.md
Lines changed: 9 additions & 0 deletions
diff --git a/‎.changeset/little-seas-lick.md
Lines changed: 41 additions & 0 deletions b/‎.changeset/little-seas-lick.md
Lines changed: 41 additions & 0 deletions
diff --git a/‎CLAUDE.md
Lines changed: 3 additions & 0 deletions b/‎CLAUDE.md
Lines changed: 3 additions & 0 deletions
diff --git a/‎examples/playground/supabase/functions/_flows/analyze_website.ts
Lines changed: 13 additions & 10 deletions b/‎examples/playground/supabase/functions/_flows/analyze_website.ts
Lines changed: 13 additions & 10 deletions
diff --git a/‎examples/playground/supabase/functions/_tasks/saveWebsite.ts
Lines changed: 9 additions & 21 deletions b/‎examples/playground/supabase/functions/_tasks/saveWebsite.ts
Lines changed: 9 additions & 21 deletions
diff --git a/‎pkgs/dsl/README.md
Lines changed: 88 additions & 7 deletions b/‎pkgs/dsl/README.md
Lines changed: 88 additions & 7 deletions
diff --git a/‎pkgs/dsl/__tests__/env-validation.test.ts
Lines changed: 108 additions & 0 deletions b/‎pkgs/dsl/__tests__/env-validation.test.ts
Lines changed: 108 additions & 0 deletions
@@ -0,0 +1,9 @@
+---
+'@pgflow/edge-worker': patch
+---
+
+Remove PlatformAdapter from EdgeWorker.start() return type
+
+The `EdgeWorker.start()` method now returns `Promise<void>` instead of `Promise<PlatformAdapter>`. The PlatformAdapter was never intended to be part of the public API and was accidentally exposed. This change properly encapsulates the internal implementation details.
+
+**Breaking Change**: If you were using the returned PlatformAdapter (which was undocumented), you'll need to refactor your code. However, this is unlikely as the EdgeWorker is designed to be started and run without further interaction.
@@ -0,0 +1,41 @@
+---
+'@pgflow/edge-worker': patch
+'@pgflow/dsl': patch
+---
+
+Add context object as second parameter to handlers
+
+Queue and flow handlers now receive an optional context parameter that provides platform resources like database connections, environment variables, and Supabase clients - eliminating boilerplate and connection management.
+
+```typescript
+// Queue handler
+EdgeWorker.start(async (payload, context) => {
+  await context.sql`INSERT INTO tasks (data) VALUES (${payload})`;
+});
+
+// Flow step handler
+.step({ slug: 'process' }, async (input, context) => {
+  const result = await context.serviceSupabase.from('users').select();
+})
+```
+
+**Core resources** (always available):
+
+- `context.env` - Environment variables
+- `context.shutdownSignal` - AbortSignal for graceful shutdown
+- `context.rawMessage` - Original pgmq message with metadata
+- `context.stepTask` - Current step task details (flow handlers only)
+
+**Supabase platform resources**:
+
+- `context.sql` - PostgreSQL client (postgres.js)
+- `context.anonSupabase` - Supabase client with anonymous key
+- `context.serviceSupabase` - Supabase client with service role key
+
+To use Supabase resources in flows, import from the Supabase preset:
+
+```typescript
+import { Flow } from '@pgflow/dsl/supabase';
+```
+
+The context parameter is optional for backward compatibility - existing single-parameter handlers continue to work unchanged.
@@ -16,3 +16,6 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 @.claude/testing_guidelines.md
 @.claude/mcp_crawler.md
 
+## Development Guidelines
+
+- Always run 'nx' as 'pnpm nx'
@@ -30,14 +30,17 @@ export default new Flow<Input>({
     const { keywords } = await extractTags(input.website.content);
     return keywords;
   })
-  .step({ slug: 'saveToDb', dependsOn: ['summary', 'tags'] }, async (input) => {
-    const websiteData = {
-      user_id: input.run.user_id,
-      website_url: input.run.url,
-      summary: input.summary,
-      tags: input.tags,
-    };
-    const { website } = await saveWebsite(websiteData);
+  .step(
+    { slug: 'saveToDb', dependsOn: ['summary', 'tags'] },
+    async (input, { serviceSupabase }) => {
+      const websiteData = {
+        user_id: input.run.user_id,
+        website_url: input.run.url,
+        summary: input.summary,
+        tags: input.tags,
+      };
+      const { website } = await saveWebsite(websiteData, serviceSupabase);
 
-    return website;
-  });
+      return website;
+    },
+  );
@@ -1,30 +1,18 @@
-import { createClient, type SupabaseClient } from '@supabase/supabase-js';
+import type { SupabaseClient } from '@supabase/supabase-js';
 import type { Database } from '../database-types.d.ts';
 
-let _supabase: SupabaseClient<Database> | undefined;
-
-function getSupabase(): SupabaseClient<Database> {
-  if (!_supabase) {
-    const SUPABASE_URL = Deno.env.get('SUPABASE_URL')!;
-    const SUPABASE_SERVICE_ROLE_KEY = Deno.env.get(
-      'SUPABASE_SERVICE_ROLE_KEY',
-    )!;
-    _supabase = createClient<Database, 'public'>(
-      SUPABASE_URL,
-      SUPABASE_SERVICE_ROLE_KEY,
-    );
-  }
-
-  return _supabase;
-}
-
-export default async (websiteData: {
+interface WebsiteData {
   user_id: string;
   website_url: string;
   summary: string;
   tags: string[];
-}) => {
-  const { data } = await getSupabase()
+}
+
+export default async (
+  websiteData: WebsiteData,
+  supabase: SupabaseClient<Database>,
+) => {
+  const { data } = await context.supabase
     .schema('public')
     .from('websites')
     .insert([websiteData])
 
@@ -13,7 +13,7 @@ Key features:
 
 - **Type Safety** - Complete TypeScript type checking from flow inputs to outputs
 - **Fluent Interface** - Chainable method calls for defining steps and dependencies
-- **Functional Approach** - Clean separation between task implementation and flow orchestration 
+- **Functional Approach** - Clean separation between task implementation and flow orchestration
 - **JSON-Compatible** - All inputs and outputs are JSON-serializable for database storage
 - **Immutable Flow Definitions** - Each step operation returns a new Flow instance
 
@@ -68,21 +68,101 @@ In pgflow, each step receives an `input` object that contains:
 2. **`input.{stepName}`** - Outputs from dependency steps
 
 This design ensures:
+
 - Original flow parameters are accessible throughout the entire flow
 - Data doesn't need to be manually forwarded through intermediate steps
 - Steps can combine original input with processed data from previous steps
 
+### Context Object
+
+Step handlers can optionally receive a second parameter - the **context object** - which provides access to platform resources and runtime information.
+
+```typescript
+.step(
+  { slug: 'saveToDb' },
+  async (input, context) => {
+    // Access platform resources through context
+    const result = await context.sql`SELECT * FROM users WHERE id = ${input.userId}`;
+    return result[0];
+  }
+)
+```
+
+#### Core Context Resources
+
+All platforms provide these core resources:
+
+- **`context.env`** - Environment variables (`Record<string, string | undefined>`)
+- **`context.shutdownSignal`** - AbortSignal for graceful shutdown handling
+- **`context.rawMessage`** - Original pgmq message with metadata
+  ```typescript
+  interface PgmqMessageRecord<T> {
+    msg_id: number;
+    read_ct: number;
+    enqueued_at: Date;
+    vt: Date;
+    message: T; // <-- this is your 'input'
+  }
+  ```
+- **`context.stepTask`** - Current step task details (flow handlers only)
+  ```typescript
+  interface StepTaskRecord<TFlow> {
+    flow_slug: string;
+    run_id: string;
+    step_slug: string;
+    input: StepInput<TFlow, StepSlug>; // <-- this is handler 'input'
+    msg_id: number;
+  }
+  ```
+
+#### Supabase Platform Resources
+
+When using the Supabase platform with EdgeWorker, additional resources are available:
+
+- **`context.sql`** - PostgreSQL client (postgres.js)
+- **`context.anonSupabase`** - Supabase client with anonymous key
+- **`context.serviceSupabase`** - Supabase client with service role key
+
+To use Supabase resources, import the `Flow` class from the Supabase preset:
+
+```typescript
+import { Flow } from '@pgflow/dsl/supabase';
+
+const MyFlow = new Flow<{ userId: string }>({
+  slug: 'my_flow',
+}).step({ slug: 'process' }, async (input, context) => {
+  // TypeScript knows context includes Supabase resources
+  const { data } = await context.serviceSupabase
+    .from('users')
+    .select('*')
+    .eq('id', input.userId);
+
+  // Use SQL directly
+  const stats = await context.sql`
+      SELECT COUNT(*) as total FROM events 
+      WHERE user_id = ${input.userId}
+    `;
+
+  return { user: data[0], eventCount: stats[0].total };
+});
+```
+
+> [!NOTE]
+> Context is optional - handlers that don't need platform resources can omit the second parameter for backward compatibility.
+
+For more details on available resources and platform configuration, see the [@pgflow/edge-worker documentation](https://github.com/pgflow-org/pgflow/tree/main/pkgs/edge-worker#context-resources).
+
 ### Flow Configuration
 
 Configure flows and steps with runtime options:
 
 ```typescript
 new Flow<Input>({
-  slug: 'my_flow',     // Required: Unique flow identifier
-  maxAttempts: 3,      // Optional: Maximum retry attempts (default: 1)
-  baseDelay: 5,        // Optional: Base delay in seconds for retries (default: 1)
-  timeout: 10,         // Optional: Task timeout in seconds (default: 30)
-})
+  slug: 'my_flow', // Required: Unique flow identifier
+  maxAttempts: 3, // Optional: Maximum retry attempts (default: 1)
+  baseDelay: 5, // Optional: Base delay in seconds for retries (default: 1)
+  timeout: 10, // Optional: Task timeout in seconds (default: 30)
+});
 ```
 
 ## Compiling Flows
@@ -120,4 +200,5 @@ Run `nx test dsl` to execute the unit tests via [Vitest](https://vitest.dev/).
 ## Documentation
 
 For detailed documentation on the Flow DSL, visit:
-- [Understanding the Flow DSL](https://pgflow.dev/explanations/flow-dsl/)
+
+- [Understanding the Flow DSL](https://pgflow.dev/explanations/flow-dsl/)
@@ -0,0 +1,108 @@
+import { describe, it, expectTypeOf } from 'vitest';
+import { Flow } from '../src/platforms/supabase.js';
+
+/**
+ * This test file verifies the environment type validation system.
+ * It tests that:
+ * 1. Supabase flows get typed environment variables
+ * 2. UserEnv augmentation works correctly
+ * 3. Invalid UserEnv types cause compilation errors
+ */
+
+describe('Environment Type Validation', () => {
+  it('should provide typed Supabase environment variables', () => {
+    const flow = new Flow({ slug: 'test_flow' })
+      .step({ slug: 'process' }, async (input, ctx) => {
+        // Supabase-specific env vars should be typed
+        expectTypeOf(ctx.env.EDGE_WORKER_DB_URL).toEqualTypeOf<string>();
+        expectTypeOf(ctx.env.EDGE_WORKER_LOG_LEVEL).toEqualTypeOf<string | undefined>();
+        
+        // Random env vars should still be accessible
+        expectTypeOf(ctx.env.RANDOM_VAR).toEqualTypeOf<string | undefined>();
+        
+        return { processed: true };
+      });
+
+    void flow;
+  });
+
+  it('should work with custom resources', () => {
+    interface CustomResources {
+      logger: { log: (msg: string) => void };
+      cache: { get: (key: string) => string | null };
+    }
+
+    const flow = new Flow<{ input: string }, CustomResources>({ slug: 'custom_flow' })
+      .step({ slug: 'process' }, async (input, ctx) => {
+        // Should have all platform resources
+        void ctx.sql;
+        void ctx.anonSupabase;
+        void ctx.serviceSupabase;
+        
+        // Plus custom resources
+        ctx.logger.log('test');
+        ctx.cache.get('key');
+        
+        // And typed env vars
+        expectTypeOf(ctx.env.EDGE_WORKER_DB_URL).toEqualTypeOf<string>();
+        
+        return { done: true };
+      });
+
+    void flow;
+  });
+
+  it('should validate that Extra extends Record<string, unknown>', () => {
+    // These should compile fine
+    type ValidExtra1 = { redis: unknown };
+    type ValidExtra2 = { logger: unknown; cache: unknown };
+    type ValidExtra3 = Record<string, unknown>;
+    
+    type TestInput = { data: string };
+    
+    // This demonstrates the constraint works
+    const flow1 = new Flow<TestInput, ValidExtra1>({ slug: 'test1' });
+    const flow2 = new Flow<TestInput, ValidExtra2>({ slug: 'test2' });
+    const flow3 = new Flow<TestInput, ValidExtra3>({ slug: 'test3' });
+    
+    // Just verify they exist
+    void flow1;
+    void flow2;
+    void flow3;
+    
+    // Note: Invalid types like string, number, etc. would cause compilation errors
+    // but we can't easily test compilation failures in vitest
+  });
+});
+
+// This demonstrates how users would augment UserEnv
+// In a real project, this would be in a separate .d.ts file
+declare module '@pgflow/dsl' {
+  interface UserEnv {
+    DATABASE_URL: string;
+    API_KEY: string;
+    OPTIONAL_FLAG?: string;
+  }
+}
+
+describe('UserEnv Augmentation', () => {
+  it('should support user-defined environment variables', () => {
+    const flow = new Flow({ slug: 'user_env_flow' })
+      .step({ slug: 'process' }, async (input, ctx) => {
+        // User-defined env vars should be typed
+        expectTypeOf(ctx.env.DATABASE_URL).toEqualTypeOf<string>();
+        expectTypeOf(ctx.env.API_KEY).toEqualTypeOf<string>();
+        expectTypeOf(ctx.env.OPTIONAL_FLAG).toEqualTypeOf<string | undefined>();
+        
+        // Platform env vars should still be there
+        expectTypeOf(ctx.env.EDGE_WORKER_DB_URL).toEqualTypeOf<string>();
+        
+        // Random vars should still work
+        expectTypeOf(ctx.env.UNKNOWN_VAR).toEqualTypeOf<string | undefined>();
+        
+        return { processed: true };
+      });
+
+    void flow;
+  });
+});