Pipelean

Sequential async pipelines with first-class retry, error boundaries, and smart failure strategies. Pragmatic, direct, no heavy abstractions.

Just plain JavaScript. Eager execution. Perfect stack traces.

Why Pipelean?

To stop writing the same try/catch and manual accumulation boilerplate.

## This is bad coding
for await (const item of iterable) {
  try {
    const result = await execute(item)
  } catch (error) {
    // OH BOY
    console.error(error)
  }
}

## This does not have async transformations and error control
array.filter(predicate).map(transform)

Pipelean gives you:

• series for sequential work over arrays and async iterables — live onProgress, pause rate limits, take, first-class error strategies
• scan / reduce for stateful accumulation across many items
• flow for stateful accumulation across one input — each operation enriches the same state
• pipe for vertical composition
• tryCatch and retry middleware you can reuse across your app
• Structured results {results, errors, sourceErrors, failure} — no silent crashes
• *Sync variants for synchronous code — same error collection, no promises

The alternatives

Need parallel? → p-map Want lazy iterators? → iter-tools Love reactive streams? → RxJS / most.js

We believe Pipelean is a pragmatic middle path: sequential by design, with built-in error control and resiliency — so you stop rewriting the same boilerplate every time.

Pipelean focuses on sequential workflows: compose operations, process collections one item at a time, carry state when needed, and control failures with built-in retry and error policies.

ESLint Plugin

Pipelean ships a small ESLint plugin that flags .forEach(), .reduce(), .map(async ...), non-generator loops, and Promise.* static combinators, suggesting pipelean equivalents. It is a separate entry point — importing it does not pull in the runtime library.

import pipeleanConfig from 'pipelean/eslint/config'

export default [
  pipeleanConfig,
  // Optionally tighten individual rules:
  {
    rules: {
      'pipelean/no-loop-without-yield': 'warn',      // loops allowed only in yielding generators
      'pipelean/no-promise-combinators': 'warn',     // suggests series() / tryCatch()
    },
  },
]

AI & Agentic Development

Pipelean is "Agent-Ready." It ships with built-in Skills to help AI assistants (like Claude, Gemini CLI, or Cursor) write better code using this library.

1. Install Skills

The easiest way to install the skills is using the Vercel agent-skills CLI:

npx skills add https://github.com/ildella/pipelean/tree/master/skills

This will install:

• pipelean-core
• pipelean-functional-programming

2. (Experimental) using skills-npm

yarn add -D skills-npm
yarn skills-npm

Documentation

• Architecture : The philosophy and design principles.
• Guide : Core concepts and usage patterns.
• Examples - Practical usage examples for all functions
• Reference - Reference docs

Example

import {pipe, series, collect} from 'pipelean'

const pipeline = pipe(
  downloadSomething,
  transformSomething,
  writeToDatabase,
)

async function* pages() {
  yield* items
}

const {results, errors, sourceErrors} = await series(pages(), pipeline, {
  strategy: collect,
  pause: 200,
  onProgress: ({item, result, index, total}) => {
    updateBar(index + 1, total)
  },
})

onProgress fires live after each kept item and is awaited before the next one. pause rate-limits. total is omitted when the source has no cheap length. See docs/patterns.md for paging, unknown length, and dead-source recipes.

Stateful accumulation: flow()

When each step should enrich the same state object (one input, many enrichments, final accumulated value), use flow():

import { flow } from 'pipelean'

const prepareAlbum = state => ({title: state.rawTitle.trim()})
const extractYear = state => ({year: parseYear(state.rawYear)})
const extractArtists = state => ({artists: state.artists ?? []})

const processAlbum = flow([
  prepareAlbum,
  extractYear,
  extractArtists,
])

const {value, errors, failure} = await processAlbum(input)
// value = {title, year, artists, ...input}

flow() defines the operation pipeline upfront and returns a function that runs that flow against different inputs. Each operation receives the current accumulated state and must return an object patch that gets shallow-merged in. Errors are handled per operation using Pipelean strategies, the same as series and scan. See docs/reference.md for the full reference.