test: add comprehensive tests for map step compilation, type constraints, and edge cases

- Introduced new integration tests for flow compilation involving map steps
- Added runtime validation tests for step dependencies and slug validation
- Included type inference validation tests for map method constraints
- Covered edge cases such as empty arrays and various runtime options
- Added tests for flow with only map steps and multiple independent chains
- Ensured correct parameter ordering and handling of dependencies in compiled SQL
- Expanded coverage for root and dependent map compilation scenarios
- Included tests for flow with only map steps and chaining behaviors
- Improved test suite robustness for map step handling in flow compilation

Author: jumski

.changeset/easy-bats-nail.md

@@ -0,0 +1,26 @@
+---
+'@pgflow/example-flows': patch
+'@pgflow/dsl': patch
+---
+
+Add `.map()` method to Flow DSL for defining map-type steps
+
+The new `.map()` method enables defining steps that process arrays element-by-element, complementing the existing SQL Core map infrastructure. Key features:
+
+- **Root maps**: Process flow input arrays directly by omitting the `array` property
+- **Dependent maps**: Process another step's array output using `array: 'stepSlug'`
+- **Type-safe**: Enforces Json-compatible types with full TypeScript inference
+- **Different handler signature**: Receives individual items `(item, context)` instead of full input object
+- **Always returns arrays**: Return type is `HandlerReturnType[]`
+- **SQL generation**: Correctly adds `step_type => 'map'` parameter to `pgflow.add_step()`
+
+Example usage:
+```typescript
+// Root map - processes array input
+new Flow<string[]>({ slug: 'process' })
+  .map({ slug: 'uppercase' }, (item) => item.toUpperCase())
+
+// Dependent map - processes another step's output
+new Flow<{}>({ slug: 'workflow' })
+  .array({ slug: 'items' }, () => [1, 2, 3])
+  .map({ slug: 'double', array: 'items' }, (n) => n * 2)

.claude/settings.json

@@ -23,6 +23,8 @@
       "Bash(grep:*)",
       "Bash(ls:*)",
       "Bash(mkdir:*)",
+      "Bash(npx tsc:*)",
+      "Bash(pnpm tsc:*)",
       "Bash(pnpm eslint:*)",
       "Bash(pnpm nx build:*)",
       "Bash(pnpm nx lint:*)",

PLAN.md

@@ -9,7 +9,7 @@
 - ✅ **DONE**: Dependency count propagation for map steps
 - ✅ **DONE**: Array element extraction - tasks receive individual array elements
 - ✅ **DONE**: Output aggregation - inline implementation aggregates map task outputs for dependents
-- ⏳ **NEXT**: DSL support for `.map()` for defining map steps
+- ✅ **DONE**: DSL support for `.map()` for defining map steps
 
 ### Chores
 
@@ -83,22 +83,24 @@
   - Validates non-array outputs to map steps fail correctly
   - Fixed broadcast aggregation to send full array not individual task output
 
-#### ❌ Remaining Work
-
-- [ ] **DSL Support for .map() Step Type**
+- [x] **PR #218: DSL Support for .map() Step Type** - `09-18-add-map-support-to-dsl` (THIS PR)
 
-  - Add `.map()` method to Flow DSL for defining map steps
+  - Added `.map()` method to Flow DSL for defining map steps
   - Constraints:
     - Locked to exactly one dependency (enforced at compile time)
     - Dependency must return an array (type-checked)
   - Syntax design:
     - Dependent maps: `flow.map({ slug: 'stepName', array: 'arrayReturningStep' }, handler)`
-    - Root maps: Decide between `{ array: 'run' }` or omitting array property
+    - Root maps: Omit array property
   - Return type always inferred as array
   - Comprehensive tests:
     - Runtime validation of array dependencies
     - Type safety for input/output types
     - Compile-time enforcement of single dependency rule
+  - Fixed complex TypeScript type inference issue with overloads
+  - Updated DSL README with .map() documentation
+
+#### ❌ Remaining Work
 
 - [ ] **Fix Orphaned Messages on Run Failure**
 
@@ -137,7 +139,7 @@
   - Ensure clean migration path from current production schema
   - If NULL improvement is done, include it in the consolidated migration
 
-- [ ] **Update README's** and **Docs**
+- [ ] **Update core README**
 
   - `pkgs/core/README.md`
 
@@ -149,12 +151,7 @@
     - Explain root map vs dependent map and how it gets handled and what restrictions those apply on the Flow input
     - Explain cascade completion of taskless steps and its limitations
 
-  - `pkgs/dsl/README.md`
-
-    - Briefly describe the new `.array()` and `.map()` methods
-    - Mention `.array` is mostly sugar, and `.map` is the new step type
-    - Link to `pkgs/core/README.md` sections for more explanation about map steps
-    - Make sure to mention that maps are constrained to have exactly one dependency
+- [ ] **Add docs page**
 
   - **Add basic docs page**
 

pkgs/dsl/README.md

@@ -73,6 +73,78 @@ This design ensures:
 - Data doesn't need to be manually forwarded through intermediate steps
 - Steps can combine original input with processed data from previous steps
 
+### Step Methods
+
+The Flow DSL provides three methods for defining steps in your workflow:
+
+#### `.step()` - Regular Steps
+
+The standard method for adding steps to a flow. Each step processes input and returns output.
+
+```typescript
+.step(
+  { slug: 'process', dependsOn: ['previous'] },
+  async (input) => {
+    // Access input.run and input.previous
+    return { result: 'processed' };
+  }
+)
+```
+
+#### `.array()` - Array-Returning Steps
+
+A semantic wrapper around `.step()` that provides type enforcement for steps that return arrays. Useful for data fetching or collection steps.
+
+```typescript
+.array(
+  { slug: 'fetch_items' },
+  async () => [1, 2, 3, 4, 5]
+)
+```
+
+#### `.map()` - Array Processing Steps
+
+Processes arrays element-by-element, similar to JavaScript's `Array.map()`. The handler receives individual items instead of the full input object.
+
+```typescript
+// Root map - processes flow input array
+new Flow<string[]>({ slug: 'process_strings' })
+  .map({ slug: 'uppercase' }, (item) => item.toUpperCase());
+
+// Dependent map - processes another step's output
+new Flow<{}>({ slug: 'data_pipeline' })
+  .array({ slug: 'numbers' }, () => [1, 2, 3])
+  .map({ slug: 'double', array: 'numbers' }, (n) => n * 2)
+  .map({ slug: 'square', array: 'double' }, (n) => n * n);
+```
+
+**Key differences from regular steps:**
+- Uses `array:` instead of `dependsOn:` for specifying the single array dependency
+- Handler signature is `(item, context) => result` instead of `(input, context) => result`
+- Return type is always an array
+- Generates SQL with `step_type => 'map'` parameter for pgflow's map processing
+
+**Type Safety:**
+The `.map()` method provides full TypeScript type inference for array elements:
+
+```typescript
+type User = { id: number; name: string };
+
+new Flow<{}>({ slug: 'user_flow' })
+  .array({ slug: 'users' }, (): User[] => [
+    { id: 1, name: 'Alice' },
+    { id: 2, name: 'Bob' }
+  ])
+  .map({ slug: 'greet', array: 'users' }, (user) => {
+    // TypeScript knows user is of type User
+    return `Hello, ${user.name} (ID: ${user.id})`;
+  });
+```
+
+**Limitations:**
+- Can only depend on a single array-returning step
+- TypeScript may not track type transformations between chained maps (use type assertions if needed)
+
 ### Context Object
 
 Step handlers can optionally receive a second parameter - the **context object** - which provides access to platform resources and runtime information.
@@ -114,6 +186,11 @@ All platforms provide these core resources:
     msg_id: number;
   }
   ```
+- **`context.workerConfig`** - Resolved worker configuration with all defaults applied
+  ```typescript
+  // Provides access to worker settings like retry limits
+  const isLastAttempt = context.rawMessage.read_ct >= context.workerConfig.retry.limit;
+  ```
 
 #### Supabase Platform Resources
 

pkgs/dsl/__tests__/integration/map-flow.test.ts

@@ -0,0 +1,208 @@
+import { describe, it, expect } from 'vitest';
+import { Flow } from '../../src/dsl.js';
+import { compileFlow } from '../../src/compile-flow.js';
+
+describe('Map flow integration tests', () => {
+  describe('complete flow examples', () => {
+    it('should compile a data processing pipeline with maps', () => {
+      // Simulating a real-world data processing flow
+      const flow = new Flow<string[]>({ slug: 'data_processing' })
+        .map({ slug: 'normalize' }, (item) => item.trim().toLowerCase())
+        .map({ slug: 'validate', array: 'normalize' }, (item) => {
+          // Validate each normalized item
+          return item.length > 0 && item.length < 100;
+        })
+        .step({ slug: 'summarize', dependsOn: ['validate'] }, (input) => ({
+          total: input.validate.length,
+          valid: input.validate.filter(v => v).length,
+          invalid: input.validate.filter(v => !v).length
+        }));
+
+      const sql = compileFlow(flow);
+
+      expect(sql).toHaveLength(4);
+      expect(sql[0]).toBe("SELECT pgflow.create_flow('data_processing');");
+      expect(sql[1]).toBe("SELECT pgflow.add_step('data_processing', 'normalize', step_type => 'map');");
+      expect(sql[2]).toBe("SELECT pgflow.add_step('data_processing', 'validate', ARRAY['normalize'], step_type => 'map');");
+      expect(sql[3]).toBe("SELECT pgflow.add_step('data_processing', 'summarize', ARRAY['validate']);");
+    });
+
+    it('should compile an ETL flow with array generation and mapping', () => {
+      const flow = new Flow<{ sourceIds: string[] }>({ slug: 'etl_flow' })
+        .array({ slug: 'fetch_data' }, async ({ run }) => {
+          // Simulating fetching data for each source ID
+          return run.sourceIds.map(id => ({ id, data: `data_${id}` }));
+        })
+        .map({ slug: 'transform', array: 'fetch_data' }, (record) => ({
+          ...record,
+          transformed: record.data.toUpperCase(),
+          timestamp: Date.now()
+        }))
+        .map({ slug: 'enrich', array: 'transform' }, async (record) => ({
+          ...record,
+          enriched: true,
+          metadata: { processedAt: new Date().toISOString() }
+        }))
+        .step({ slug: 'load', dependsOn: ['enrich'] }, async (input) => {
+          // Final loading step
+          return {
+            recordsProcessed: input.enrich.length,
+            success: true
+          };
+        });
+
+      const sql = compileFlow(flow);
+
+      expect(sql).toHaveLength(5);
+      expect(sql[1]).not.toContain("step_type"); // array step
+      expect(sql[2]).toContain("step_type => 'map'");
+      expect(sql[3]).toContain("step_type => 'map'");
+      expect(sql[4]).not.toContain("step_type"); // regular step
+    });
+
+    it('should handle complex nested array processing', () => {
+      // Flow that processes nested arrays (e.g., matrix operations)
+      const flow = new Flow<number[][]>({ slug: 'matrix_flow' })
+        .map({ slug: 'row_sums' }, (row) => row.reduce((a, b) => a + b, 0))
+        .step({ slug: 'total_sum', dependsOn: ['row_sums'] }, (input) =>
+          input.row_sums.reduce((a, b) => a + b, 0)
+        );
+
+      const sql = compileFlow(flow);
+
+      expect(sql).toHaveLength(3);
+      expect(sql[1]).toBe("SELECT pgflow.add_step('matrix_flow', 'row_sums', step_type => 'map');");
+      expect(sql[2]).toBe("SELECT pgflow.add_step('matrix_flow', 'total_sum', ARRAY['row_sums']);");
+    });
+  });
+
+  describe('runtime validation', () => {
+    it('should throw when trying to use non-existent step as array dependency', () => {
+      const flow = new Flow<{}>({ slug: 'test' })
+        .step({ slug: 'exists' }, () => [1, 2, 3]);
+
+      expect(() => {
+        // @ts-expect-error - TypeScript should catch this at compile time
+        flow.map({ slug: 'fail', array: 'doesNotExist' }, (item) => item);
+      }).toThrow('Step "fail" depends on undefined step "doesNotExist"');
+    });
+
+    it('should throw when step slug already exists', () => {
+      const flow = new Flow<number[]>({ slug: 'test' })
+        .map({ slug: 'process' }, (n) => n * 2);
+
+      expect(() => {
+        flow.map({ slug: 'process' }, (n) => n * 3);
+      }).toThrow('Step "process" already exists in flow "test"');
+    });
+
+    it('should validate slug format', () => {
+      expect(() => {
+        new Flow<number[]>({ slug: 'test' })
+          .map({ slug: 'invalid-slug!' }, (n) => n);
+      }).toThrow(); // validateSlug should reject invalid characters
+    });
+
+    it('should validate runtime options', () => {
+      // This should not throw - valid options
+      const validFlow = new Flow<number[]>({ slug: 'test' })
+        .map({
+          slug: 'valid',
+          maxAttempts: 3,
+          baseDelay: 1000,
+          timeout: 30000,
+          startDelay: 5000
+        }, (n) => n);
+
+      expect(compileFlow(validFlow)).toHaveLength(2);
+
+      // Invalid options should be caught by validateRuntimeOptions
+      expect(() => {
+        new Flow<number[]>({ slug: 'test' })
+          .map({
+            slug: 'invalid',
+            maxAttempts: 0 // Should be >= 1
+          }, (n) => n);
+      }).toThrow();
+    });
+  });
+
+  describe('type inference validation', () => {
+    it('should correctly infer types through map chains', () => {
+      const flow = new Flow<{ items: string[] }>({ slug: 'test' })
+        .step({ slug: 'extract', dependsOn: [] }, ({ run }) => run.items)
+        .map({ slug: 'lengths', array: 'extract' }, (item) => item.length)
+        .map({ slug: 'doubles', array: 'lengths' }, (len) => len * 2)
+        .step({ slug: 'sum', dependsOn: ['doubles'] }, (input) => {
+          // Type checking - this should compile without errors
+          const total: number = input.doubles.reduce((a, b) => a + b, 0);
+          return total;
+        });
+
+      const sql = compileFlow(flow);
+      expect(sql).toHaveLength(5);
+    });
+  });
+
+  describe('edge cases', () => {
+    it('should handle empty array processing', () => {
+      const flow = new Flow<Json[]>({ slug: 'empty_test' })
+        .map({ slug: 'process' }, (item) => ({ processed: item }));
+
+      const sql = compileFlow(flow);
+      expect(sql).toHaveLength(2);
+      expect(sql[1]).toContain("step_type => 'map'");
+    });
+
+    it('should handle all runtime options combinations', () => {
+      const flow = new Flow<string[]>({ slug: 'options_test' })
+        .map({ slug: 'no_options' }, (s) => s)
+        .map({ slug: 'some_options', array: 'no_options', maxAttempts: 5 }, (s) => s)
+        .map({
+          slug: 'all_options',
+          array: 'some_options',
+          maxAttempts: 3,
+          baseDelay: 1000,
+          timeout: 30000,
+          startDelay: 5000
+        }, (s) => s);
+
+      const sql = compileFlow(flow);
+
+      expect(sql[1]).toBe("SELECT pgflow.add_step('options_test', 'no_options', step_type => 'map');");
+      expect(sql[2]).toBe("SELECT pgflow.add_step('options_test', 'some_options', ARRAY['no_options'], max_attempts => 5, step_type => 'map');");
+      expect(sql[3]).toContain("max_attempts => 3");
+      expect(sql[3]).toContain("base_delay => 1000");
+      expect(sql[3]).toContain("timeout => 30000");
+      expect(sql[3]).toContain("start_delay => 5000");
+      expect(sql[3]).toContain("step_type => 'map'");
+    });
+
+    it('should handle map steps with no further dependencies', () => {
+      // Map step as a leaf node
+      const flow = new Flow<number[]>({ slug: 'leaf_map' })
+        .map({ slug: 'final_map' }, (n) => n * n);
+
+      const sql = compileFlow(flow);
+      expect(sql).toHaveLength(2);
+      expect(sql[1]).toBe("SELECT pgflow.add_step('leaf_map', 'final_map', step_type => 'map');");
+    });
+
+    it('should handle multiple independent map chains', () => {
+      const flow = new Flow<{ a: number[]; b: string[] }>({ slug: 'parallel' })
+        .step({ slug: 'extract_a' }, ({ run }) => run.a)
+        .step({ slug: 'extract_b' }, ({ run }) => run.b)
+        .map({ slug: 'process_a', array: 'extract_a' }, (n) => n * 2)
+        .map({ slug: 'process_b', array: 'extract_b' }, (s) => s.toUpperCase())
+        .step({ slug: 'combine', dependsOn: ['process_a', 'process_b'] }, (input) => ({
+          numbers: input.process_a,
+          strings: input.process_b
+        }));
+
+      const sql = compileFlow(flow);
+      expect(sql).toHaveLength(6);
+      expect(sql[3]).toContain("step_type => 'map'");
+      expect(sql[4]).toContain("step_type => 'map'");
+    });
+  });
+});