# SuperPipe

[![CI status][ci-img]][ci-url]
[![Coverage][coverage-img]][coverage-url]
[![License MIT][license-img]][license-url]
[![JavaScript Style Guide][standard-img]][standard-url]
[![NPM version][npm-img]][npm-url]

A lightweight functional reactive programming (FRP) library for composing asynchronous operations with dependency injection.

## Installation

```sh
npm install superpipe
```

## Quick Start

```javascript
import superpipe from 'superpipe'

// Create a pipeline factory with dependencies
const sp = superpipe({
  greet: name => `Hello, ${name}!`
})

// Build a pipeline
const pipeline = sp('greeting-pipeline')
  .input(['name'])
  .pipe('greet', 'name', 'message')
  .pipe(message => console.log(message), 'message')
  .end()

// Execute the pipeline
pipeline('World')  // Output: Hello, World!
```

## Core Concepts

### Pipelines

A pipeline is a sequence of pipes that execute in order. Each pipe can:
- Transform data
- Produce outputs that become available to subsequent pipes
- Control flow through the declarative boolean channels (raw boolean
  dependencies, `!`-prefixed pipes)

### Pipes

Each pipe has three components:
- **Function**: The operation to perform (can be a function or a string referencing an injected dependency)
- **Input**: Dependencies the function needs (retrieved from the store or injected deps)
- **Output**: Names to assign to the return values

### Dependency Injection

Dependencies are passed when creating the superpipe factory and are available to all pipelines:

```javascript
const sp = superpipe({
  db: databaseConnection,
  logger: loggingService,
  config: appConfig
})
```

## API

### `superpipe(deps?)`

Creates a pipeline factory function.

- `deps` (optional): Object containing dependencies available to all pipelines

Returns a function `(name, defs?) => PipelineAPI | executor`

### Pipeline API

#### `.input(names)`

Maps positional arguments to named dependencies.

```javascript
sp('my-pipeline')
  .input(['userId', 'action'])  // First arg -> userId, second -> action
```

#### `.pipe(fn, input?, output?)`

Adds a pipe to the pipeline.

- `fn`: Function to execute, or string name of an injected dependency
- `input`: String or array of dependency names to pass as arguments
- `output`: String or array of names to assign to return values

```javascript
// Direct function
.pipe((a, b) => a + b, ['x', 'y'], 'sum')

// Injected function by name
.pipe('myFunction', ['arg1', 'arg2'], 'result')

// Using the special `next` input — next(error, value) assigns `value`
// to the pipe's declared output name
.pipe((next, value) => {
  setTimeout(() => next(null, value * 2), 100)
}, ['next', 'value'], 'key')
```

#### Output binding

The output spec decides how a pipe's return value is stored — the same
spec always means the same thing, whatever the return's type:

| Spec | Meaning |
| --- | --- |
| `'out'` | bind the whole return value under `out` |
| `'{a, b}'` | pick the named properties |
| `['a', 'b']` | destructure — positional for array returns, by name for objects |
| `'{...}'` | merge every key of the returned object |
| *(none)* | effects only — the return value is discarded |

```javascript
.pipe(() => ({ id: 1, name: 'ada' }), null, 'user')   // user === the whole object
.pipe(() => ({ id: 1, name: 'ada' }), null, '{name}')  // name === 'ada'
.pipe(() => ['a', 'b'], null, ['first', 'second'])     // positional
.pipe(() => ({ ok: true }), null, '{...}')             // merges ok into the store
.pipe(save)                                            // side effects only
```

A pipe with no output spec discards its return value — declare an output
(or `'{...}'`) when a pipe produces data. The `...` marker only works as
the entire spec — mixing it with names (`'{a, ...}'`) is rejected at
construction.

Destructure specs validate what they name: every key a `'{a, b}'` pick
(or an array spec, or a `source:destination` rename) names must exist on
the returned object, a positional spec must not exceed an array return,
and a destructure spec receiving a return it cannot destructure (a
primitive, `null`, a function) — or no return at all, including a
promise that resolves to nothing — is a spec/return mismatch. Every one of
these throws `OutputKeyError` — a typo like `{reolvedTarget}` fails at
the pipe that produced it, not as a silent `undefined` three pipes
later. A key that exists with value `undefined` is fine: presence, not
truthiness, is the contract. `{a, b}` picks by `in`, so prototype-
inherited keys count; `'{...}'` merges only the returned object's own
enumerable keys (`Object.keys`), so inherited keys are left out. Values
delivered alongside an error (`next(error, partialValue)`) merge without
validation — shape mismatches included — so a failing pipe's partial
result never masks the real error on its way to the error handler.

#### Business-result early return

Use `result:<name>` as a stage's output spec when it returns a business result
rather than an ordinary value. `{ value: value }` stores `value` as `<name>` and
continues. `{ reason: reason }` stores `reason` as `<name>` and successfully
stops the remaining stages. The result object must contain exactly one of those
keys. This is opt-in: ordinary returned objects, including `{ error: ... }`,
remain data. Thrown values and `next(error)` remain pipeline failures.
Existing `result:destination` rename mappings continue to work when the returned
object supplies its legacy `result` property.

```javascript
const findUser = superpipe({ lookup, loadProfile })('find user')
  .input('id')
  .pipe(lookup, 'id', 'result:user')
  .pipe(loadProfile, 'user', 'profile')
  .end('user')

findUser('missing') // the reason returned by lookup; loadProfile does not run
```

#### `.onExit(handler)` and `.reason(handler)`

A pipeline can end four ways, and until now three of them were silent: every
exit that was not a natural completion simply settled with whatever the
container held. `.onExit(handler)` registers an observer that runs exactly once
when a run settles, however it ended.

The handler receives an exit record and the container:

| field | meaning |
| --- | --- |
| `via` | `'value'`, `'reason'`, `'halt'`, `'error'` or `'abort'` |
| `step` | index of the stage that ended the run; `null` when no single stage did (natural completion, abort) |
| `name` | that stage's function name, under the same rule |
| `reason` | the rejected value, on a `'reason'` exit only |
| `error` | the failure the caller sees, when there is one — usually an `'error'` or `'abort'` exit, but also a `'halt'` or `'reason'` exit that a later abort ended; `null` otherwise. Typed `unknown`, since a pipe may throw or reject with any value |

```javascript
const sp = superpipe({ authorize, capture })
const run = sp('charge')
  .input('order')
  .pipe(authorize, 'order', 'result:authorization')
  .pipe(capture, 'authorization', 'receipt')
  .onExit((exit, container) => audit.write(exit))
  .end('receipt')
```

Exit handlers observe a run; they do not handle it. They run before the error
handler and before an `endAsync` promise resolves, in registration order, and a
handler that throws is contained — it can never change the run's outcome, and
the thrown value is discarded rather than reported.

The container a handler receives is a snapshot of the run's named outputs
without the pipeline's own `next` continuation, so a handler observes the run
rather than being able to re-enter it.

Four details worth knowing. A run cancelled by an already-aborted signal never
starts, so its handlers receive an `'abort'` exit and an **empty** container —
there is no run state to report; correlate on the exit record rather than on
container fields if you need to cover that case. `via: 'value'` means every
stage ran; with `endAsync`, fetching the requested output happens after the run
settles, so a malformed output spec can still reject the promise after handlers
have seen a `'value'` exit. A halted run with continuations still
outstanding fires its handlers when the last one drains, not at the moment it
halted. And
the first exit recorded wins, so a run that halts and is then aborted reports
`via: 'halt'` while its `endAsync` promise still rejects with
`PipelineAbortedError` — `via` names what ended the stages, `error` what the
caller sees.

`.reason(handler)` is the rejection counterpart to `.error(handler)`. A typed
rejection is an expected business outcome rather than a failure, so it settles
the run quietly; this gives it the same first-class handling an error has. Only
one reason handler is allowed per pipeline, and it fires only on a `'reason'`
exit.

```javascript
.reason((reason, exit) => metrics.increment(`denied.${exit.name}`))
```

#### `.error(handler, input?)`

Sets an error handler for the pipeline. Only one error handler is allowed per pipeline.

```javascript
.error((error) => console.error('Pipeline error:', error), 'error')
```

#### `.end(output?)`

Finalizes the pipeline and returns an executor function. When `output` is
provided, the executor returns the requested output value (or an object of the
requested keys) once the pipeline completes; otherwise it returns `undefined`.

```javascript
const run = sp('my-pipeline')
  .input(['x'])
  .pipe(x => x * 2, 'x', 'doubled')
  .end('doubled')

run(5)  // Executes the pipeline with x=5, returns 10
```

### Declarative API

Pipelines can also be defined declaratively:

```javascript
const run = sp('math-pipeline', [
  ['input', ['a', 'b']],
  [(a, b) => a + b, ['a', 'b'], 'sum'],
  [(sum) => console.log('Sum:', sum), 'sum']
])

run(3, 4)  // Output: Sum: 7
```

## Special Features

### Special Input Dependencies

- `next`: Control when to proceed to the next pipe and assign outputs manually
  (for async operations)

```javascript
// Async operation — next(error, value) proceeds and stores `value`
// under the pipe's declared output name ('data')
.pipe((next, value) => {
  fetchData(value, (err, result) => {
    next(err, result)
  })
}, ['next', 'value'], 'data')
```

### Boolean Flow Control

`false` steers the pipeline only on the **declarative channels** — a raw
boolean dependency or a `!`-prefixed injected pipe. A function pipe's
return value, `false` included, is ordinary data: it is stored under the
output name and the pipeline continues.

```javascript
const sp = superpipe({ isBlocked: false })

// Raw boolean dependency — flow control: falsey halts
sp('guard')
  .pipe('isBlocked', 'user')  // Halts unless isBlocked is truthy
  .end()

// `!`-prefixed injected pipe — inverts the boolean for flow control
sp('not-blocked')
  .pipe('!isBlocked', 'user')  // Continues only while isBlocked is falsey
  .end()

// Function pipe — the boolean return is DATA, the pipeline continues
sp('admin-check')
  .pipe((user) => user.isAdmin, 'user', 'isAdmin')
  .end()
```

### Not Pipes (`!`)

Prefix function name with `!` to invert boolean results:

```javascript
.pipe('!isBlocked', 'user')  // Continues if isBlocked returns false
```

### Optional Pipes (`?`)

Prefix function name with `?` to skip if the dependency is undefined:

```javascript
.pipe('?optionalHandler', 'maybeValue')  // Skips if optionalHandler or maybeValue is undefined
```

### Output Renaming (`source:destination`)

A `source:destination` spec picks the returned object's `source` property
and stores it as `destination` — a one-key form of brace picking:

```javascript
.pipe(getData, 'id', 'result:userProfile')  // Stores the returned `result` as `userProfile`
```

### Reserved Output Names and Shadowing

The runtime container reserves `next` as a control name. A pipe output —
declared, renamed, or merged from a returned plain object — or an invocation
input that writes a reserved name throws. A pipe output whose name matches a
configured dependency also throws, since it would silently shadow it:

```javascript
superpipe({ shared: fn })
// throws on execution:
.pipe(() => 'value', null, 'shared')
```

Invocation inputs are the exception: a caller may deliberately override a
configured dependency by passing a per-run value under the same name.

### Asynchronous Pipelines and `.end(output)`

A pipe that does not request `next` may return a Promise (or any thenable):
resolution continues the pipeline with the value, rejection triggers the error
handler. Returning a thenable from a pipe that declares `next` throws — pick
one continuation channel.

```javascript
sp('async-pipeline')
  .pipe(() => repository.getWorkflow(), null, 'workflow')  // Promise-returning
  .pipe((workflow) => render(workflow), 'workflow')
  .error((error) => console.error('Failed:', error.message), 'error')
  .end()
```

The executor returned by `.end()` completes synchronously. When a pipe uses
`next` or returns a Promise, values produced later are not reflected in
`.end(output)`'s return value — use `.endAsync(output)` for that:

```javascript
const run = sp('fetch-workflow')
  .pipe(() => repository.getWorkflow(), null, 'workflow')
  .error((error) => console.error('Failed:', error.message), 'error')
  .endAsync('workflow')

const workflow = await run()   // Promise — settles when the run settles
```

`.endAsync` settles when the *run* settles — every pipe executed, a
flow-control halt fired, or an error was dispatched. A halted run resolves
with the partial snapshot; a failed run rejects with the active error even
when an error handler ran (the promise is an additional observer). Fully
synchronous pipelines resolve immediately, so `await` works uniformly.
Alternatively, deliver async results through a final pipe, `next`, or an
error handler.

The runner returned by `.endAsync` is reusable — call it as many times as
you like. Cancellation is per run: pass an `AbortSignal` to the runner's
`.withSignal(signal, ...args)` method for a single cancellable invocation.
An aborted run rejects with `PipelineAbortedError` — its `name` is
`AbortError`, and its `reason` carries the signal's abort reason — without
invoking the error handler. A signal already aborted at call time rejects
before the first pipe runs:

```javascript
import superpipe, { PipelineAbortedError } from 'superpipe'

const run = sp('fetch-workflow')
  .pipe(() => repository.getWorkflow(), null, 'workflow')
  .endAsync('workflow')

const controller = new AbortController()
const promise = run.withSignal(controller.signal)   // one run, one signal
controller.abort()   // elsewhere / on cancel: rejects the pending run

try {
  const workflow = await promise
  // ... use workflow
} catch (error) {
  if (!(error instanceof PipelineAbortedError)) {
    throw error   // a real pipeline failure, not a cancellation
  }
  // cancelled — nothing to do
}

// The runner itself is unaffected by the abort — keep using it:
const again = await run.withSignal(new AbortController().signal)
const plain = await run()   // no cancellation
```

Cancellation stops the run, not just the caller's view of it: when the
signal aborts, no pipe that has not started will execute, every live
`next` callback is disabled (a retained callback becomes a no-op and
releases its hold on the run's state), and the returned promise rejects
immediately with `PipelineAbortedError` — the cancellation itself never
reaches the pipeline's error handler. Continuations already in flight are
discarded when they land, errors included.

An operation already in flight is not preempted — JavaScript cannot
interrupt a running function — so a pipe whose operation ignores the
signal may still finish that operation; its result is discarded and no
downstream pipe runs. Pass the same signal into the underlying operations
(fetch, model calls) so they stop early too.

A run counts as completed only once the returned promise settles: a
successful run defers its settlement by one job (so an error dispatched in
the same unwind wins), which means an abort fired synchronously right
after `run.withSignal(...)` returns — before any `await` — still cancels a
run whose pipes all finished in that tick.

### Object-String Syntax

Use an `{a, b}` object string to destructure a single object argument into
several inputs, or to pick specific keys from a pipe's returned object as
outputs:

```javascript
// Input: pull `arg1` and `arg2` out of the single object argument
sp('my-pipeline')
  .input('{arg1, arg2}')
  .pipe(({ arg1, arg2 }) => arg1 + arg2, '{arg1, arg2}', 'sum')

// Output: pick keys from the returned object
.pipe(getProfile, 'id', '{name, email}')

// Output: merge every key of the returned object into the store
.pipe(getProfile, 'id', '{...}')
```

## Error Handling

Errors can be triggered by:
1. Calling `next(error)` with an error
2. Throwing an exception in a pipe function

The active error travels on the pipeline's execution state, not the data
container: a pipe result or output named `error` is ordinary data and does
not trigger the error handler. The handler's `error` input always receives
the active failure.

```javascript
sp('safe-pipeline')
  .input(['data'])
  .pipe((data) => {
    if (!data) throw new Error('Data required')
    return data
  }, 'data', 'validated')
  .error((error) => {
    console.error('Error:', error.message)
  }, 'error')
  .end()
```

## TypeScript Support

SuperPipe is written in TypeScript and includes type definitions. Values
flowing through a pipeline (`PipeResult`, `PipeOutput`) are typed
`unknown` — the executor narrows them; input/output specs accept the
`source:destination` rename form (`PipeRename`):

```typescript
import superpipe, { Dependencies, PipelineAPI } from 'superpipe'

interface MyDeps extends Dependencies {
  logger: (msg: string) => void
}

const sp = superpipe<MyDeps>({
  logger: console.log
})
```

### Typed pipelines

`typed(deps)` is a second entry point over the same runtime that carries the
pipeline's context through the type system, so the compiler checks how stages
connect and the final runner needs no cast. The untyped API is unchanged.

```typescript
import { typed, type Result } from 'superpipe'

const tp = typed({ loadUser, isBlocked: false })

const findUser = tp<[string]>('find-user')
  .input('id')
  .pipe(admit, 'id', 'result:user')          // admit: (id: string) => Result<User, Denied>
  .pipe((user) => user.name, 'user', 'name')  // user: User
  .reason((reason) => audit(reason))          // reason: Denied
  .endAsync('user')                           // (id: string) => Promise<User | Denied>
```

What the types track:

- `tp<[A, B]>(name)` declares the invocation arguments; `.input(['a', 'b'])`
  binds them positionally and the runner's parameters are inferred from them.
- `.pipe(fn, ['a', 'b'], 'out')` checks that every input name exists (context
  first, then deps), that `fn` accepts those values positionally, and binds
  `out` to the awaited return. Array outputs destructure (positional for tuple
  returns, by name for object returns) and `'{...}'` merges the returned
  object. `'{...}'` is typed for object-literal returns: the runtime merges
  only enumerable own keys, and the types reject functions, arrays, known
  built-ins such as `Date` and `Map`, and shapes whose every member is a
  method, but a class instance with data fields cannot be told from a
  literal, so its methods would be typed yet absent. Brace picks and
  `src:dst` renames are not typed yet.
- `'result:<name>'` requires the pipe to return `Result<T, R>`: `<name>` is
  `T` inside the pipeline, and `R` joins the union the `.reason()` handler
  receives.
- `.end(name)` and `.endAsync(name)` return the value bound to `name`, plus the
  reason types of every gate whose output is `result:<name>`, plus `undefined`
  only when an earlier gate, boolean halt, or optional skip can leave `name`
  unset. Ending a pipeline on its gate's own name therefore gives `T | R` with
  no runtime change.
- Injected pipes (`'loadUser'`, `'!isBlocked'`, `'?enrich'`) resolve the
  dependency's type: functions have their parameters checked against the
  declared inputs, boolean deps and `!` pipes are flow control, and `?` pipes
  make their output possibly `undefined`.
- Output names that are reserved (`next`) or shadow a configured dependency are
  compile errors, matching the runtime `OutputNameError`. Problems like these
  are reported through the builder's return type as `TypedError<message>`, so
  the error appears at the next chained call with the message in it.
- Once any pipe returns a promise or declares `next`, `.end(name)` is refused
  at compile time; use `.endAsync(name)`.
- The runtime's ordering rules are compile errors too: `.input()` after the
  first pipe, `.pipe()` or a second `.error()` after the error handler, and a
  second `.reason()`.
- A pipe that declares `next` binds its output as `unknown`.

The typed builder uses `const` type parameters, so consumers need TypeScript
5.0 or later.

## Go Port

SuperPipe also ships as a Go module — same semantics, idiomatic Go: one
blocking `Run`, typed spec constructors instead of the string mini-grammar,
`context.Context` cancellation, and `errors.Is`-checkable sentinels. See
[`go/README.md`](go/README.md) for the user guide and
[`docs/go-port-spec.md`](docs/go-port-spec.md) for the full contract.

## License

[MIT](LICENSE)

## Bug Reports

[GitHub Issues](https://github.com/lsm/superpipe/issues)

[license-img]: https://img.shields.io/npm/l/superpipe.svg
[license-url]: http://opensource.org/licenses/MIT
[npm-img]: http://img.shields.io/npm/v/superpipe.svg
[npm-url]: https://npmjs.org/package/superpipe
[ci-img]: https://github.com/lsm/superpipe/actions/workflows/ci.yml/badge.svg
[ci-url]: https://github.com/lsm/superpipe/actions/workflows/ci.yml
[coverage-img]: https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/lsm/superpipe/badges/coverage-badge.json
[coverage-url]: https://github.com/lsm/superpipe/actions/workflows/ci.yml
[standard-img]: https://img.shields.io/badge/code_style-standard-brightgreen.svg
[standard-url]: https://standardjs.com