SuperPipe

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

Installation

npm install superpipe

Quick Start

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:

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.

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

// 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 |

.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.

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 |

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.

.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.

.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.

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:

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)

// 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.

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:

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

Optional Pipes (?)

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

.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:

.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:

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.

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:

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:

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:

// 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.

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):

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.

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 for the user guide and docs/go-port-spec.md for the full contract.

License

MIT

Bug Reports

GitHub Issues