feat: Add .map() Method to Flow DSL for Map Step Support

[jumski]

Overview

This PR introduces the .map() method to the Flow DSL, enabling users to define map-type steps that process arrays element-by-element. This completes the DSL support for the map step infrastructure that was previously added to the SQL Core layer.

Problem

Previously, while the SQL Core supported map steps (with step_type => 'map'), there was no way to define them via the TypeScript DSL. Users had to manually write SQL or work around this limitation.

Solution

The new .map() method provides a type-safe, intuitive way to define map steps with these key characteristics:

API Design

// Root map - processes flow input array
new Flow<string[]>({ slug: 'text_processing' })
  .map({ slug: 'normalize' }, (text) => text.trim().toLowerCase())

// Dependent map - processes another step's array output
new Flow<{}>({ slug: 'workflow' })
  .array({ slug: 'fetch_items' }, () => fetchData())
  .map({ slug: 'enrich', array: 'fetch_items' }, (item) => ({
    ...item,
    enriched: true
  }))

Key Features

1. Intuitive dependency syntax
   • Omit array property for root maps (processes flow input)
   • Use array: 'stepSlug' for dependent maps (single dependency only)
2. Different handler signature
   • Map handler: (item, context) => Json
   • Regular step handler: (input, context) => Json
   • Map handlers receive individual array elements, not the full input object
3. Type safety
   • Enforces Json-compatible types for inputs and outputs
   • Full TypeScript inference for item types based on dependencies
   • Compile-time validation that dependencies return arrays
4. SQL generation
   • Correctly adds step_type => 'map' parameter to pgflow.add_step()
   • Maintains proper parameter ordering: dependencies, options, then step_type

Implementation Details

Changes Made

1. Extended StepDefinition interface (dsl.ts)
   • Added optional stepType?: 'single' | 'map' property
   • Defaults to 'single' for backward compatibility
2. Implemented .map() method (dsl.ts)
   • Complex type signatures for proper inference
   • Runtime validation for dependencies
   • Stores stepType: 'map' in step definition
3. Updated compile function (compile-flow.ts)
   • Checks for stepType property
   • Adds step_type => 'map' parameter when needed
4. Comprehensive test coverage
   • Type tests: 20 tests covering constraints and inference
   • Runtime tests: 12 tests for SQL generation
   • Integration tests: 12 tests for complete workflows
   • Example flows demonstrating various use cases

Technical Decisions

• Optional array key: Mirrors the dependsOn pattern (no key = root step)
• Single dependency constraint: Enforced at both type and runtime levels
• Handler signature difference: Clear distinction between map and regular steps
• Type parameter ordering: Worked around TypeScript constraints for optional parameters

Testing

The implementation follows TDD principles with:

• ✅ Type safety tests (inference, constraints, edge cases)
• ✅ SQL generation tests (parameter ordering, step_type inclusion)
• ✅ Integration tests (complete workflows, validation)
• ✅ Example flows (data processing, ETL, parallel maps)

Migration Guide

No breaking changes. Existing flows continue to work unchanged. To use map steps:

1. For root maps (array input):
2. For dependent maps:

Future Considerations

• The input.run convention will be removed in future versions
• Map steps are ready for this transition (handlers don't use input.run)
• Type system properly handles both root and dependent map cases

Checklist

• Implementation complete with type safety
• SQL generation working correctly
• Comprehensive test coverage
• Example flows created
• No breaking changes
• Documentation in changeset

Related Issues

Part of the map infrastructure implementation series (see PLAN.md)

[changeset-bot]

🦋 Changeset detected

Latest commit: 53cc331

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 7 packages

Name	Type
@pgflow/dsl	Patch
@pgflow/client	Patch
@pgflow/core	Patch
@pgflow/edge-worker	Patch
@pgflow/example-flows	Patch
pgflow	Patch
@pgflow/website	Patch

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

✨ Finishing touches

🧪 Generate unit tests

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch 09-18-add-map-support-to-dsl

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[jumski]

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

• test: add integration tests for flows with map steps #222
• feat: add task_index attribute to step_task_record and start_tasks function #221
• Fix Orphaned Messages on Run Failure #220
• chore: improve verify-schemas-synced and regenerate-temp-migration #219
• feat: Add .map() Method to Flow DSL for Map Step Support #218 👈 (View in Graphite)
• feat: add Map Step Output Aggregation #217
• feat: enhance start_tasks with conditional input logic and add array handling migration #216
• chore: make initial_tasks NULL by default to indicate 'unknown yet' #215
• feat: implement cascade completion of taskless steps #214
• # 🚀 Task Spawning for Map Steps #213
• feat: add root map step support #209
• feat: add map step type in sql #208
• feat: add map and array steps for parallel processing #207
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[nx-cloud]

View your CI Pipeline Execution ↗ for commit 53cc331

Command	Status	Duration	Result
nx run-many -t build --projects client,dsl --co...	✅ Succeeded	6s	View ↗
nx affected -t build --configuration=production...	✅ Succeeded	4s	View ↗
nx affected -t lint typecheck test --parallel -...	✅ Succeeded	5m 55s	View ↗

---

☁️ Nx Cloud last updated this comment at 2025-09-18 10:16:25 UTC

[github-actions]

🔍 Preview Deployment: Website

✅ Deployment successful!

🔗 Preview URL: https://pr-218.pgflow.pages.dev

📝 Details:

• Branch: 09-18-add-map-support-to-dsl
• Commit: 4c21b84c72b650ecf3dfd4e94b7af539c3e297a4
• View Logs

_Last updated: _

[github-actions]

🔍 Preview Deployment: Playground

✅ Deployment successful!

🔗 Preview URL: https://pr-218--pgflow-demo.netlify.app

📝 Details:

• Branch: 09-18-add-map-support-to-dsl
• Commit: 4c21b84c72b650ecf3dfd4e94b7af539c3e297a4
• View Logs

_Last updated: _