# Pipelean Reference Documentation

**Overview**

`functional.js` is an async programming library that provides core utilities for handling asynchronous operations, error management, data transformations, and functional composition patterns in JavaScript. The library is designed with a pragmatic philosophy: avoid heavy abstractions, use eager execution, and provide clear, predictable error handling.

**Key Principles**

- **Pragmatic**: Plain JavaScript, eager execution, sequential processing — plus `join()` for concurrent fork/join over named tasks.
- **First class error handling**: Multiple error strategies for different use cases

---

## Table of Contents

### Error strategies

- [failFast](#failfast) - Error Strategy Identifier (aliases: `stopOnError`, `fail`)
- [rethrow](#rethrow) - Error Strategy Identifier (throws on first error)
- [failLate](#faillate) - Error Strategy Identifier
- [collect](#collect) - Error Strategy Identifier
- [skip](#skip) - Error Strategy Identifier
- [Callbacks](#callbacks) - Error handling callbacks

### Iterators

**Horizontal**

- [series](#series) - Stateless Sequential Execution
- [scan](#scan) - Stateful Sequential Transformation
- [reduce](#reduce) - Pure Reduction
- [filter](#filter) - Stateless Selection
- [stopWhen](#stopwhen) - Source Adapter for Predicate-Based Early Exit (alias of sorts: the `while` you don't need)
- [forever](#forever) - Source Adapter for Unbounded Ticking (the `while (true)` you don't need)

### Composition

- [pipe](#pipe) - Vertical Composition
- [flow](#flow) - Stateful Accumulation Across One Input
- [join](#join) - Fork/join over Named Concurrent Tasks

### Misc

- [assign](#assign) - Conditional property assignment for `flow`
- [where](#where) - Object-pattern predicate
- [retry](#retry) - Configurable Retry Logic
- [tryCatch](#trycatch) - Single Function Lifecycle Hooks

### Sync Variants

- [Sync Functions](#sync-functions) - Synchronous counterparts for synchronous code

---

## Error strategies

### failFast

**Purpose**: Error strategy identifier that stops immediately on the first error.

**Alias**: `stopOnError`

**Use when**: Critical operations where failure means the entire pipeline is invalid.

**Behavior**:
- Sets `failure: {item, error, index}` on first error
- Calls `onError({item, error, index, total})`, then `onFailure({item, error, index})` immediately
- Stops iteration

**Return Format**: `{results: [], errors: [], failure: {item, error, index}}`

**Example**:
```javascript
import {series, failFast} from 'pipelean'

const result = await series([1, 2, 3], async item => {
  if (item === 2) throw new Error('Error')
  return item * 2
}, {strategy: failFast})

// result = {results: [], errors: [], failure: {item: 2, error: Error(...), index: 1}}
```

---

### rethrow

**Purpose**: Error strategy identifier that throws the error immediately on the first failure. Does NOT return a structured result on failure.

**Use when**: "Let it crash" / fail-early patterns where the caller is expected to handle errors externally (e.g. via try/catch).

**Behavior**:
- Throws the error on first failure
- Does NOT call `onFailure`
- Does NOT call `onError` in `series`

**Return Format**: On success: `{results, errors: [], failure: false}`

**Example**:
```javascript
import {series, rethrow} from 'pipelean'

try {
  const result = await series([1, 2, 3], async item => {
    if (item === 2) throw new Error('Error')
    return item * 2
  }, {strategy: rethrow})
} catch (error) {
  // error = Error('Error')
}
```

---

### collect

**Purpose**: Error strategy identifier that collects all errors and continues processing (default for `series` and `filter`).

**Use when**: Batch operations, logging scenarios, background tasks.

**Behavior**:
- Collects all errors in `errors` array
- Sets `failure: false`
- Does NOT call `onFailure`

**Return Format**: `{results, errors: [{item, error, index}], failure: false}`

**Example**:
```javascript
import {series, collect} from 'pipelean'

const result = await series([1, 2, 3], async item => {
  if (item === 2 || item === 4) throw new Error('Error')
  return item * 2
}, {strategy: collect})

// result = {results: [2, 6], errors: [{item: 2, error: ..., index: 1}], failure: false}
```
---

### failLate

**Purpose**: Error strategy identifier that collects all errors and returns `failure: {errors}` at the end.

**Use when**: Application-layer needs to detect if *any* error occurred.

**Behavior**:
- Collects all errors in `errors` array
- Sets `failure: {errors}` after loop completes (only if `errors.length > 0`)
- Calls `onFailure({errors})` if `failure` is truthy

**Return Format**: `{results, errors: [{item, error, index}], failure: {errors}}` (if any errors occurred)

**Example**:
```javascript
import {series, failLate} from 'pipelean'

const result = await series([1, 2, 3], async item => {
  if (item === 2 || item === 4) throw new Error('Error')
  return item * 2
}, {strategy: failLate})

// result = {results: [2, 6], errors: [{item: 2, error: ..., index: 1}], failure: {errors}}
```
---

### skip

**Purpose**: Error strategy identifier that ignores errors entirely (no collection), but `onError` is still called if present.

**Use when**: Best-effort processing, some failures are acceptable.

**Behavior**:
- Ignores errors (no collection, `errors` stays empty)
- Sets `failure: false`
- Does NOT call `onFailure`

**Return Format**: `{results, errors: [], failure: false}`

**Example**:
```javascript
import {series, skip} from 'pipelean'

const result = await series([1, 2, 3], async item => {
  if (item === 2) throw new Error('Error')
  return item * 2
}, {strategy: skip})

// result = {results: [2, 6], errors: [], failure: false}
```

---

## Callbacks

### onError

Optional callback for verification/telemetry (logging, metrics).

- Called for every handled item error in `series`
- Does NOT affect control flow
- Receives `{item, error, index, total}`; `total` is omitted when unknown
- Use for: logging, metrics, external error reporting

```javascript
await series(items, fn, {
  strategy: skip,
  onError: ({item, error}) => console.error('Error:', item.id, error.message)
})
```

---

### onFailure

Optional callback for application-layer error handling (UI updates, notifications).

- Called when `failure` is truthy
- Depends on strategy:
  - `failFast`: called with `{item, error, index}`
  - `failLate`: called with `{errors}`
  - `collect` / `skip`: NOT called (failure is false)

```javascript
await series(items, fn, {
  strategy: failLate,
  onFailure: (failure) => {
    if (failure.errors) {
      showToast('Some items failed')
    } else {
      showToast(`Item ${failure.item} failed: ${failure.error.message}`)
    }
  }
})
```

---

## Iterators

### series

**Purpose**: Stateless sequential execution over an iterable. Runs a function on each item, one at a time.

**Type**: `(items, fn, opts?) => Promise<Outcome>` — immediate mode
Curried: `(fn, opts?) => (items) => Promise<Outcome>`

**Parameters**:
- `items`: An iterable (array, async iterable, generator)
- `fn(item, index)`: The function to apply. Returns the mapped value, or throws, or returns `undefined` to drop the item.
- `opts` (optional):
  - `strategy`: Error strategy (default: `collect`). `failFast`, `collect`, `failLate`, `skip`, `rethrow`.
  - `onProgress({item, result, index, total})`: Called after each successful item. NOT called for errors or `undefined` drops.
  - `onError({item, error, index, total})`: Called for each handled item error. Does not affect control flow.
  - `onFailure(failure)`: Called when failure is truthy. Receives `{item, error, index}` for `failFast`, `{errors}` for `failLate`.
  - `onSourceError({error, index})`: Called when the iteration itself throws (a source error). Never called under `rethrow`; mutually exclusive with `onError`.
  - `take`: Limit items processed.
  - `total`: Explicit planned input count for progress/error callbacks. If omitted, `series` uses a cheap known input size when available. If `take` is set, callback `total` is limited to `Math.min(take, knownTotal)`. If no total is known, the `total` key is omitted.
  - `pause`: Milliseconds between successful items.
  - `pauseOnErrors`: Whether to also pause after errors (default: `false`).

**Return**: `{results, errors, sourceErrors, failure}` where `failure` is `false` on success. Collected `errors` are `{item, error, index}`. `sourceErrors` is an array of source-error contexts, empty when the source completed cleanly (see [Source errors](#source-errors)).

**Usage Example**:
```javascript
import { series, failFast } from 'pipelean'

// Basic: double each number
const { results } = await series([1, 2, 3], x => x * 2)
// results = [2, 4, 6]

// With failFast: stop on first error
const result = await series(items, fn, { strategy: failFast })

// With pause for rate limiting
const result = await series(endpoints, fn, { pause: 500 })

// With UI progress
await series(items, saveItem, {
  onProgress: ({index, total}) => updateProgress(index + 1, total),
  onError: ({item, error}) => reportItemError(item.id, error),
})

// Curried: create a reusable transform
const double = series(x => x * 2)
const { results } = await double([1, 2, 3])

// Integration with pipe for filtering
import { pipe } from 'pipelean'
const { results } = await series(items, pipe(
  x => x.active ? x : undefined,  // drop inactive
  x => x.name,                     // extract name
))
```

#### Source errors

A **source error** is an error thrown by the iteration itself — the `for await` step (e.g. an async generator dying mid-stream because a DB connection was lost). This is distinct from an **operation error**, which is your `fn` throwing on an item.

A dead source is terminal: no more items can arrive, so iteration ends under every strategy. The error never reaches `onError`.

Every return path gains one additive field:

- `sourceErrors`: array of `{error, index}` contexts — no `item` key, because a source error has no item. `index` is the iteration index of the failed `.next()` call (a generator that yields 2 items then throws produces `index: 2`). Empty array when the source completed cleanly.

New option:

- `onSourceError({error, index})`: telemetry callback with the same contract as `onError`. Never called under `rethrow`. `onError` and `onSourceError` are mutually exclusive per error.

Per-strategy behavior when the source dies:

| Strategy | Behavior |
| --- | --- |
| `collect` | Results preserved, context recorded in `sourceErrors`, `failure: false` |
| `failFast` | `failure` = the source context, results cleared, accumulator (`value` in reduce mode) preserved |
| `failLate` | Results preserved, source contexts merged into end-of-run aggregation: `failure = {errors: [...errors, ...sourceErrors]}` |
| `skip` | Source errors ignored entirely: `sourceErrors: []`, results preserved, `failure: false` |
| `rethrow` | Throws immediately; no callbacks called |

```javascript
async function* pages(db) {
  while (true)
    yield await db.nextPage() // throws when the connection drops
}

const result = await series(pages(db), processPage, {
  strategy: collect,
  onSourceError: ({error, index}) => log.warn('source died', {error, index}),
})
// result.results holds every page processed before the drop
// result.sourceErrors = [{error, index}]
```

Cleanup follows native generator semantics: the generator's own `finally` runs during propagation, early exit via `take` still closes the iterator through `iterator.return()`, and if a `finally` itself throws, that error supersedes the original source error.

---

### scan

**Purpose**: Stateful sequential transformation - transforms each item and accumulates results.

**Type**: `(iterable, scanner, initialValue, opts?) => Promise<Outcome>`

**Parameters**:
- `iterable`: An async iterable (array, generator, or any object implementing the iteration protocol)
- `scanner`: A function with signature `(accumulator, item, index) => newAccumulator`
- `initialValue`: The starting value for the accumulator
- `opts` (optional):
  - `strategy`: Error strategy (default: `failFast`). `failFast`, `collect`, `failLate`, `skip`, `rethrow`.
  - `onError`: Called for each handled item error. Does not affect control flow.
  - `onFailure`: Called when failure is truthy.
  - `onSourceError`: Called when the iterable itself throws.
  - `storePartialResults`: `true` (default) returns `{results, …}`; `false` returns `{value, …}` (see `reduce`).
  - `scan` does **not** support `onProgress`, `pause`, or `take`.

**Return Type**: A Promise that resolves to an object containing:
- `results`: Array of intermediate results (or `[]` on failFast failure)
- `errors`: Array of errors encountered (empty for failFast, skip, throw)
- `sourceErrors`: Array of `{error, index}` contexts for errors thrown by the iteration itself; empty when the source completed cleanly
- `failure`: `false` on success; `{item, error, index}` for failFast; `{errors}` for failLate (source errors are merged into the aggregated errors array)
- `value`: Final accumulated value when `storePartialResults: false`. Preserved under every strategy except `rethrow`, even when the source dies mid-stream

**Source errors**: when the iterable itself throws mid-stream (e.g. an async generator dying), the accumulator and partial results collected so far are preserved per strategy and the context is reported through `onSourceError` instead of `onError`. See [Source errors](#source-errors) under `series` for the full per-strategy behavior.

**Key Characteristics**:
- **Stateful**: Each transformation depends on the previous result
- **Accumulates**: Both successful results and errors for inspection
- **Index Tracking**: Provides index of each item for correlation

**Usage Example**:
```javascript
import { scan } from './functional.js'

// Track insertions in a database
const { results, errors } = await scan(
  records,
  async (acc, record) => {
    const inserted = await db.insert(record)
    return acc + inserted  // Accumulate count
  },
  0  // Initial count
)
```

---

### reduce

**Purpose**: Pure reduction - transforms each item sequentially and returns only the final accumulated value. A convenience wrapper around `scan` with `storePartialResults: false` baked in.

**Type**: `(iterable, scanner, initialValue, opts?) => Promise<Outcome>`

**Parameters**:
- `iterable`: An async iterable (array, generator, or any object implementing the iteration protocol)
- `scanner`: A function with signature `(accumulator, item, index) => newAccumulator`
- `initialValue`: The starting value for the accumulator
- `opts` (optional):
  - `strategy`: Error strategy (default: `failFast`). `failFast`, `collect`, `failLate`, `skip`, `rethrow`.
  - `onError`: Called for each handled item error. Does not affect control flow.
  - `onFailure`: Called when failure is truthy.
  - `onSourceError`: Called when the iterable itself throws.
  - `reduce` does **not** support `onProgress`, `pause`, or `take`.

**Return Type**: A Promise that resolves to an object containing:
- `value`: The final accumulated value. Preserved under every strategy except `rethrow`, even when the source dies mid-stream
- `errors`: Array of errors encountered (empty for failFast, skip, throw)
- `sourceErrors`: Array of `{error, index}` contexts for errors thrown by the iteration itself; empty when the source completed cleanly
- `failure`: `false` on success; `{item, error, index}` for failFast; `{errors}` for failLate

**Key Characteristics**:
- **Single value**: Returns `{value, errors, failure}` instead of intermediate results
- **No boilerplate**: No need for `results.at(-1) || 0` — the final value is direct
- **Error handling**: Same error strategies as `scan` and `series`

**Usage Example**:
```javascript
import { reduce } from 'pipelean'

// Sum track durations into album total
const {value: totalDuration} = await reduce(
  tracks,
  (accumulator, {duration}) => accumulator + duration,
  0,
)
// totalDuration = 22 (no intermediate array, no .at(-1))
```

**When to use**:
- Summing, counting, concatenating
- Any reduction where you only need the final result
- When `scan`'s intermediate results array is unnecessary overhead

**When to use `scan` instead**:
- When you need every intermediate accumulator value (e.g. dependent sequential operations where each step's output is meaningful)

**Alias**: `scanReduce` is kept as an alias of `reduce` for backward compatibility and points to the same function.

---

### filter

**Purpose**: Stateless selection tool - filters items from an iterable based on a predicate function. Delegates to `series` internally: the predicate is converted to a transform that returns the original item (keep) or `undefined` (drop).

**Type**: `(...args) => Promise<Outcome> | filterFunction`

**Parameters**:
- `predicate`: A function `(item, index) => truthy | falsy`, or a plain object pattern (converted via `where()`)
- `items`: The iterable to filter (immediate mode)
- `opts` (optional): Options passed through to `series`

**Options**: Same as `series` — `strategy`, `onError`, `onFailure`, `take`, `total`, `onProgress`, `pause`, `pauseOnErrors`.

**Return Type**: `{ results, errors, failure }` — same shape as `series`:
- `results`: Original items where the predicate returned truthy
- `failure`: `false` on success (no errors), `{item, error, index}` for `failFast`, `{errors}` for `failLate`

**Key Characteristics**:
- The predicate's return value is never placed into `results` — only truthiness is checked, and the original `item` is what gets kept or dropped.
- Pattern objects are supported: `filter(users, {active: true})` works via `where()`. Anything implementing the iteration protocols (arrays, generators, source adapters like `stopWhen`) is always treated as a source, never as a pattern.

**Usage Example**:
```javascript
import { filter } from 'pipelean'

const adults = await filter(
  users,
  user => user.age >= 18,
)
// result.results = [user1, user3, ...] — original items, not predicate output
```

---

### stopWhen

**Purpose**: Source adapter that stops pulling from an iterable as soon as a predicate is truthy. The standard way to cancel or predicate-stop any source consumed by `series`, `scan`, `reduce`, `filter`, or a raw `for await`.

**Type**: `(items, predicate?) => asyncIterable`
Sync: `(items, predicate?) => iterable`

**Parameters**:
- `items`: An array or (async) iterable — generators and paging sources included
- `predicate(item, index)`: Checked **before** each item is yielded. Truthy → stop. Defaults to `() => false` (never stops). Sync only.

**Key Characteristics**:
- **Check before yield**: the triggering item is pulled from the source but never offered downstream — "cancel before work"
- **Pull-lazy**: once stopped, the underlying source is abandoned mid-stream; native cleanup (`iterator.return()`, generator `finally`) still runs
- **Clean completion, not an error**: consumers see a finished source — `failure: false`, empty `sourceErrors`. Cancel is a shorter run, not a source death
- **Composable**: zero changes in `series` / `scan` / `reduce` / `filter`; they just see an iterable that ends early. Works with raw `for await` too
- **Length forwarding**: when the source has a numeric `length` (arrays), it is forwarded on the wrapper so `series` keeps progress totals without an explicit `total`. Generators have no cheap size — `total` stays omitted
- **Predicate throws are source errors**: reported through `onSourceError` / `sourceErrors`, never `onError`
- This is **not** intra-item abort: work already started on an item is not interrupted (use `AbortController` in your operation for that)

**Usage Example**:
```javascript
import { series, stopWhen } from 'pipelean'

const {results} = await series(enrichOne, {
  total: albums.length,
  pause: ENRICH_DELAY_MS,
  onProgress: onItem,
})(stopWhen(albums, shouldStop))

// limit-style: close over counters instead of the item
stopWhen(folders, () => count >= limit)
```

---

### forever

**Purpose**: Source adapter that yields the natural numbers 0, 1, 2, ... with no end — the unbounded counterpart of `stopWhen`. The building block for daemon loops: repeat a step until an abort predicate says so. Pipelean does not ship a daemon primitive; it ships the two halves, `forever` and `stopWhen`, and the composition is the whole story.

**Type**: `() => asyncIterable<number>`
Sync: `() => iterable<number>`

**Parameters**: none.

**Key Characteristics**:
- **Unbounded**: yields forever until the consumer stops pulling — every consumption must be bounded by `stopWhen`, `take`, or a `break`. A bare `collect(forever())` never ends
- **Composable**: `stopWhen(forever(), predicate)` is a predicate-stopped tick source; the predicate (usually an abort-signal check) is checked before each tick, so "cancel before work" holds
- **Index semantics**: the yielded numbers are the tick indices a consumer receives — a cycle counter for free
- **Non-accumulating consumers recommended**: a daemon has no items, so `reduce`/`scan` (stateful, no `results` array) fit better than `series`, which would otherwise rely on the `undefined` drop signal every tick

**Usage Example**:
```javascript
import { reduce, rethrow, forever, stopWhen } from 'pipelean'

// daemon loop: one pass immediately, then repeat until Ctrl-C
const { value: progress } = await reduce(
  stopWhen(forever(), () => controller.signal.aborted),
  async (state, index) => {
    await runRound({ index })
    await sleep({ ms: everyMs, signal: controller.signal })
    return state
  },
  { cycles: 0, failures: 0 },
  { strategy: rethrow },
)
```

---

## Composition

### pipe

**Purpose**: Pipelean operation composer - chains functions left-to-right and preserves Pipelean's `undefined` drop signal.

**Type**: `(...fns) => (input) => Promise<ReturnType<LastFn>>`

**Parameters**:
- Variadic arguments: Any number of sync or async functions to execute sequentially
- `input`: The initial value passed to the first function

**Return Type**: A Promise that resolves to the final result.

**Key Characteristics**:
- Functions execute **left-to-right** (first argument is applied to `input`)
- Output of one function becomes input to the next
- Supports both synchronous and asynchronous functions
- Natural data flow from input through transformations
- Designed for composing reusable operations passed to `series()` or used directly
- **Undefined Short-Circuit**: If any step returns `undefined`, remaining steps are skipped and `undefined` is returned. This enables selection within a composed operation — see [series](#series) drop behavior.

**Usage Example**:
```javascript
import { pipe } from './functional.js'

const normalizeUser = pipe(
  async id => validateUserId(id),
  async id => fetchUser(id),
  user => user.active ? user : undefined,
  user => ({...user, email: user.email.toLowerCase()}),
)

const user = await normalizeUser(userId)

// Compose operations in a readable pipeline
const result = await pipe(
  async data => validate(data),
  async data => transform(data),
  async data => persist(data),
)(input)
```

**Best Practice**: Use `pipe()` when you need to chain operations that form a coherent data processing pipeline.

**Selection in pipe** (via undefined short-circuit):
```javascript
import { pipe, series } from 'pipelean'

// Merge filter and transform in a single operation
const result = await series(items, pipe(
  x => x.active ? x : undefined,  // drop inactive items
  x => x.name,                     // extract name
))
// Items where active is false are skipped entirely
```

---

### flow

**Purpose**: Sequentially apply multiple state-enrichment operations to one accumulated value. Each successful operation returns an object patch. Errors are handled per operation using Pipelean strategies.

**Type**: `(operations, options?) => (initialState) => Promise<{value, errors, failure}>`

**Parameters**:
- `operations`: An array of functions. Each function `operation(state)` must return a non-null, non-array object (the patch to merge into the state). Operations can be sync or async.
- `options` (optional):
  - `strategy`: Error strategy (default: `collect`). `failFast`, `collect`, `failLate`, `skip`, `rethrow`.
  - `onError({operation, error, index, total})`: Called for each handled operation error. Receives a normalized shape where `operation` is the function's `name` (or `operation-${index}` for anonymous functions) and `total` is the operation count.
  - `onFailure(failure)`: Called when `failure` is truthy. Receives `{operation, error, index}` for `failFast` and `{errors}` for `failLate` — both normalized.

**Return**: A function `(initialState) => Promise<{value, errors, failure}>`.

The returned function:
- Validates `initialState` (non-null, non-array object) and throws `TypeError` otherwise.
- Executes each operation in order, passing the **current accumulated state** to each one.
- Shallow-merges the returned patch into the state.
- Treats an invalid operation return (null, array, primitive) as that operation throwing a `TypeError`, so the selected strategy applies.
- Returns `{value, errors, failure}` where:
  - `value`: The final accumulated state. Even under `failFast`, this is the last successful accumulated state.
  - `errors`: Array of `{operation, error, index}` (normalized). Empty for `skip` and `failFast`.
  - `failure`: `false` for `collect` and `skip`; `{operation, error, index}` for `failFast`; `{errors}` for `failLate`. For `rethrow`, the original error is thrown instead of returning a result.

**Key Characteristics**:
- **Stateful accumulation**: Each operation enriches the same state object. Output of one operation becomes the input to the next.
- **Shallow merge**: `{...state, ...patch}` — top-level keys are overwritten, nested objects are not deep-merged.
- **Reusable pipeline**: The pipeline is defined once. The returned function is called with different inputs.
- **Same error strategies as `series` / `scan`**: All four strategies work, with the same `onError` / `onFailure` semantics. Default is `collect`.
- **No intermediate states**: Only the final accumulated value is returned. There is no `results` array.
- **Empty operations array** is valid and returns the initial state untouched.

**Usage Example**:
```javascript
import { flow, collect, failFast } 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, rawTitle, rawYear, ...input}

// Strategy and callbacks are bound when the pipeline is defined
const processOrStop = flow(
  [prepareAlbum, extractYear, extractArtists],
  {
    strategy: failFast,
    onError: ({operation, error, index, total}) =>
      logger.error({operation, error, index, total}),
  },
)

const result = await processOrStop(input)

// Reuse the same pipeline with different inputs
const album1 = await processAlbum(rawAlbum1)
const album2 = await processAlbum(rawAlbum2)
```

**When to use `flow()` vs `pipe()` vs `series()`**:
- `flow()`: One input, many enrichments, final accumulated value. Stateful across the pipeline.
- `pipe()`: Value-in / value-out composition. Each step transforms the previous step's return value. Stateless. Composable into `series()` and `flow()`.
- `series()`: Many inputs, one operation per input. Horizontal.

**Error callback payload** (always normalized):
- `onError({operation, error, index, total})` — `operation` is the function's `name` if set, otherwise `operation-${index}`. `total` is the operations array length.
- `onFailure({operation, error, index})` for `failFast`.
- `onFailure({errors})` for `failLate`, where each entry is `{operation, error, index}`.

---

### join

**Purpose**: Fork/join over a record of **named** async tasks. Every branch starts immediately and `join` waits for all of them to settle, returning one structured result. Errors are data, per branch, using the same strategies as `flow`. This is not a concurrency runtime: no cancellation, no scheduler, no ordering guarantee.

**Type**: `(tasks, options?) => Promise<{value, errors, failure}>`

**Parameters**:
- `tasks`: A non-null, non-array object mapping a branch **name** to a function `task(key, index)`. Each function can be sync or async. The name is the stable key for that branch in the result.
- `options` (optional):
  - `strategy`: Error strategy (default: `collect`). `collect`, `failLate`, `skip`, `rethrow`, and `failFast`.
  - `onError({operation, error, index})`: Called for each handled branch error. `operation` is the branch name.
  - `onFailure(failure)`: Called (and awaited) when `failure` is truthy. Receives `{errors}` for `failLate` / `failFast`.

**Return**: `Promise<{value, errors, failure}>`:
- `value`: Object mapping the **successful** branch names to their resolved values, in declaration order. A branch resolving `undefined` is still a success (`value[key] === undefined`).
- `errors`: Array of `{operation, error, index}`, ordered by declaration (not completion). Empty for `skip` (and for `rethrow`, which throws instead).
- `failure`: `false` for `collect` and `skip`; `{errors}` for `failLate` and `failFast`. For `rethrow`, the original error of the first failed branch is thrown instead of returning a result.

**Throws (before any branch starts)**: `TypeError` if `tasks` is not a non-null, non-array object, if any value is not a function, or if it is a thenable (e.g. an un-awaited `Promise`).

**Strategies**:
- `collect` (default): keep successes in `value`, failures in `errors`, `failure: false`.
- `failLate`: like `collect`, then `failure: {errors}` if anything failed.
- `failFast` (alias `fail`, `stopOnError`): behaves as `failLate`. Branches already started and cannot be stopped without an `AbortSignal`.
- `skip`: `errors` stays empty, but `onError` still fires.
- `rethrow`: throw the original error of the first failed branch, by declaration order, once every branch has settled.

**Key Characteristics**:
- **Named, heterogeneous branches**: A record, not a collection. Each branch has a stable key, even when it fails: the failure is reported under `errors` with that `operation`.
- **All branches settle**: `join` starts from a resolved promise, so a synchronous throw becomes a rejection; `allSettled` always captures it and nothing is left unhandled.
- **Completion order is irrelevant**: `value` keys follow declaration order, and so does `errors`.
- **No drop signal**: `undefined` is a legitimate branch value, unlike in `series` / `pipe`.
- **Empty record** is valid: resolves to `{value: {}, errors: [], failure: false}`.
- **Not a concurrent map**: for a homogeneous collection use p-map.

**Usage Example**:
```javascript
import { join, failLate } from 'pipelean'

const ping = host => fetch(`https://${host}/health`).then(r => r.status)
const whoami = () => fetch('/whoami').then(r => r.json())

const { value, errors, failure } = await join({
  api: () => ping('api.example.com'),
  cdn: () => ping('cdn.example.com'),
  session: () => whoami(),
})

// value  = { api: 200, cdn: 200, session: {...} }
// errors = [] ; failure = false

// A failed branch stays named and leaves the rest intact
const down = await join({
  api: () => ping('api.example.com'),
  cdn: () => { throw new Error('unreachable') },
})
// down.value  = { api: 200 }
// down.errors = [{ operation: 'cdn', error: Error('unreachable'), index: 1 }]

// Treat any branch failure as a failure of the join
const strict = await join(
  { api: () => ping('api.example.com') },
  { strategy: failLate },
)
```

**When to use `join()` vs `series()`**:
- `join()`: A handful of named, heterogeneous tasks that run at the same time. One structured outcome, errors per branch.
- `series()`: Many homogeneous items processed one at a time. Horizontal, ordered, with `pause` / `take` / `onProgress`.

---

## Misc

### assign

**Purpose**: Creates a `flow()` operation that conditionally assigns a property to state. If `parse(state)` returns `undefined`, the property is skipped (empty patch). Otherwise the property is set to the parsed value.

**Type**: `(property, parse) => (state) => {} | {[property]: value}`

**Parameters**:
- `property`: The property name (string) to set on state.
- `parse`: A function `(state) => value | undefined`. Return `undefined` to skip the property.

**Return Type**: A function suitable as a `flow()` operation. Returns `{}` when the parse is skipped, or `{[property]: value}` to merge into the accumulated state.

**Key Characteristics**:
- **Conditional assignment**: Only sets the property when `parse` returns a non-undefined value.
- **Skip sentinel**: Only `undefined` triggers skip — `null`, `false`, `0`, `""` are all assigned normally.
- **Composable**: Multiple `assign()` calls compose naturally inside `flow()`.

**Usage Example**:
```javascript
import { assign, flow } from 'pipelean'

const extractName = assign('name', state => state.rawName.trim())
const extractYear = assign('year', state => {
  const n = Number.parseInt(state.rawYear, 10)
  return Number.isNaN(n) ? undefined : n
})

const {value} = await flow([extractName, extractYear])({
  rawName: '  Alice  ',
  rawYear: '1995',
})
// value = {rawName: '  Alice  ', rawYear: '1995', name: 'Alice', year: 1995}
```

---

### where

**Purpose**: Creates a predicate function for strict equality object matching. Primarily used with `filter` for pattern-based selection.

**Type**: `(pattern) => (item) => boolean`

**Parameters**:
- `pattern`: A plain object with key-value pairs to match against.

**Return Type**: A predicate function.

**Usage Example**:
```javascript
import { where, filter } from 'pipelean'

const isAdmin = where({role: 'admin'})
const admins = await filter(users, isAdmin)

// Or use pattern directly with filter
const adults = await filter(users, {active: true})
```

### retry

**Purpose**: Retry async functions with configurable attempts and delays between attempts.

**Type**: `(fn, options) => retryFunction`

**Parameters**:
- `fn`: The async function to retry (required)
- `options`: Configuration object with the following properties:
  - `attempts`: `number` (default: `3`) - Number of retry attempts
  - `delay`: `number` (default: `0`) - Delay between retry attempts in milliseconds

**Behavior**:
- Retries on each attempt until successful or exhausted
- Throws the last error after exhausting all attempts
- No delay before first attempt
- Applies configured delay between subsequent attempts

**Usage Example**:
```javascript
import {retry} from 'pipelean'

const flakyOperation = async () => {
  if (Math.random() > 0.5)
    throw new Error('transient')
  return 'ok'
}

const robust = retry(flakyOperation, {attempts: 5, delay: 1000})
const result = await robust()
```

---

### tryCatch

**Purpose**: Wraps individual async functions with lifecycle hooks for comprehensive error handling and telemetry. Always catches errors — returns `null` on failure.

**Type**: `(fn, options) => wrapperFunction`

**Parameters**:
- `fn`: The async function to wrap
- `options`: Configuration object with the following properties:
  - `onStart: () => void` — Called before function execution (no arguments)
  - `onSuccess: (result) => void | Promise<void>` — Called on successful completion (result only)
  - `onError: (error) => void` — Called on error (error only)
  - `onFinally: () => void` — Called regardless of success/failure (no arguments)

**Return Type**: Returns a wrapper function with the same signature as `fn`. Returns `null` on error.

**Features**:
- Automatic `async/await` wrapping
- Preserves original function signature
- Comprehensive lifecycle: onStart → onSuccess/onError → onFinally
- Deep telemetry support for debugging

**Usage Example**:
```javascript
import { tryCatch } from 'pipelean'

const safeFetch = tryCatch(
  async (url) => {
    const response = await fetch(url)
    return await response.json()
  },
  {
    onError: (error) => {
      console.error('Fetch failed:', error)
    }
  }
)

const data = await safeFetch('https://api.example.com')
// data is the parsed JSON on success, null on error
```

---

## Sync Functions

`seriesSync`, `filterSync`, `findSync`, `scanSync`, `reduceSync`, `pipeSync`,
`flowSync`, and `tryCatchSync` are synchronous counterparts of the core
functions.

They use the same error strategies and structured return conventions as the
async iteration functions, but return values directly instead of wrapped in a
Promise. Use them when your data and operations are synchronous and you still
want Pipelean's structured error collection.

**Available sync variants**:

- `seriesSync` returns `{results, errors, sourceErrors, failure}` directly
- `filterSync` returns `{results, errors, sourceErrors, failure}` directly
- `findSync` returns `{result, errors, failure}` directly and stops at the first match
- `scanSync` returns `{results, errors, sourceErrors, failure}` directly
- `reduceSync` returns `{value, errors, sourceErrors, failure}` directly
- `pipeSync` composes synchronous functions left-to-right
- `flowSync` returns `{value, errors, failure}` directly and runs a state-enrichment pipeline synchronously
- `tryCatchSync` wraps a synchronous function with lifecycle hooks
- `stopWhenSync` is the sync source adapter — same contract as `stopWhen`, sync iterables only

The sync variants handle **source errors** with the same semantics as their async twins: `seriesSync`, `filterSync`, `scanSync`, and `reduceSync` accept `onSourceError({error, index})` and report errors thrown by the iteration itself in the `sourceErrors` field. See [Source errors](#source-errors) under `series`.

> `scanReduceSync` is kept as an alias of `reduceSync` for backward compatibility.

**Not supported in sync variants**:

- `pause` and `pauseOnErrors`, because they depend on async `delay()`
- async iterables, because sync variants use plain synchronous iteration

**Usage Example**:

```javascript
import { seriesSync, collect } from 'pipelean'

const {results, errors, failure} = seriesSync([1, 2, 3], x => {
  if (x === 2)
    throw new Error('bang')
  return x * 10
}, {strategy: collect})

// results: [10, 30]
// errors: [{item: 2, error: Error('bang'), index: 1}]
// failure: false
```

**Early-exit selection with `findSync`**:

```javascript
import { findSync } from 'pipelean'

const users = [
  {id: 1, active: false},
  {id: 2, active: true},
  {id: 3, active: true},
]

const {result, errors, failure} = findSync(users, {active: true})

// result: {id: 2, active: true}
// errors: []
// failure: false
```

`findSync` supports the same predicate forms as `filterSync`, including object
patterns via `where()`. It exits as soon as the first match is found. When no
item matches, it returns `{result: undefined, errors: [], failure: false}`.

**Synchronous state enrichment with `flowSync`**:

```javascript
import { flowSync, collect } from 'pipelean'

const processAlbumSync = flowSync([
  state => ({title: state.rawTitle.trim()}),
  state => ({year: parseYear(state.rawYear)}),
])

const {value, errors, failure} = processAlbumSync(input)
// value: the shallow-merged final state
// errors: []
// failure: false
```

`flowSync` mirrors `flow()` exactly but executes synchronously. Same
strategies, same `onError` / `onFailure` semantics, same normalized error
shapes. Use it when your operations and inputs are synchronous.

---

## Related Documentation

- **[README.md](../README.md)** - Project overview and philosophy
- **[guide.md](../guide.md)** - Comprehensive development guide with examples
- **[examples.md](../examples.md)** - Practical usage examples for all functions