Pipeline

Pipeline

Zero-dependency, TypeScript-first functional data pipeline system for composing map, filter, and reduce transformations with full type inference across sync and async operations.

Import

import { pipe, map, filter, reduce } from "bytekit/pipeline";
// or from the root entry:
import { pipe, map, filter, reduce } from "bytekit";

What it does

The pipeline system lets you compose typed transformation operators into a reusable, lazy, immutable pipeline. Nothing executes until you call .process(data). Each .pipe() call returns a new Pipeline instance — the original is never mutated.

Operator types:

• map — transform each element concurrently (Promise.all)
• filter — retain elements concurrently, preserve order
• reduce — accumulate sequentially (each step awaits the previous)

PipelineOp<TIn, TOut>

type PipelineOp<TIn, TOut> = (input: TIn) => TOut | Promise<TOut>;

The atomic unit. A function that transforms a value — synchronously or asynchronously.

Pipeline<TIn, TOut>

Method	Signature	Description
pipe(op)	(op: PipelineOp<TOut, TNext>) => Pipeline<TIn, TNext>	Append an operator. Returns a new Pipeline.
process(data)	(data: TIn) => Promise<TOut>	Execute all operators sequentially. Always async.

pipe(...ops)

Factory function that builds a Pipeline from a sequence of operators. Provides full type inference for up to 7 chained operators.

function pipe<T, A>(op1: PipelineOp<T, A>): Pipeline<T, A>;
function pipe<T, A, B>(op1: PipelineOp<T, A>, op2: PipelineOp<A, B>): Pipeline<T, B>;
// ... up to 7 ops
// Escape hatch (no inference):
function pipe<T = unknown>(...ops: PipelineOp<unknown, unknown>[]): Pipeline<T, unknown>;

map<T, U>(fn)

function map<T, U>(
  fn: (item: T, index: number) => U | Promise<U>
): PipelineOp<T[], U[]>

Parameter	Description
fn	Called with (item, index). Sync or async.
Returns	PipelineOp<T[], U[]> — all items processed concurrently via Promise.all.

filter<T>(fn)

function filter<T>(
  fn: (item: T, index: number) => boolean | Promise<boolean>
): PipelineOp<T[], T[]>

Parameter	Description
fn	Predicate receiving (item, index). Sync or async.
Returns	PipelineOp<T[], T[]> — all predicates run concurrently; retained items preserve original order.

reduce<T, U>(fn, initial)

function reduce<T, U>(
  fn: (acc: U, item: T, index: number) => U | Promise<U>,
  initial: U
): PipelineOp<T[], U>

Parameter	Description
fn	Reducer receiving (accumulator, item, index). Sync or async.
initial	Starting accumulator value.
Returns	PipelineOp<T[], U> — runs sequentially. Returns initial for empty arrays.

How It Works

1. pipe(...ops) creates a Pipeline storing the operators in an array. Nothing runs.
2. .pipe(op) clones the operator array and returns a new Pipeline. Original is unchanged.
3. .process(data) runs each operator in a for…of loop with await. The output of each op becomes the input of the next.
4. map/filter run each item's function concurrently with Promise.all; reduce runs each step sequentially.
5. Errors thrown by any operator propagate unchanged out of .process().

Examples

Sync composition

import { pipe, filter, map, reduce } from "bytekit/pipeline";

const totalRevenue = await pipe(
  filter<Product>((p) => p.inStock),
  map<Product, number>((p) => p.priceCents),
  reduce<number, number>((acc, price) => acc + price, 0)
).process(products);

Async map (concurrent enrichment)

import { pipe, map } from "bytekit/pipeline";

const enriched = await pipe(
  map<Order, EnrichedOrder>(async (order) => ({
    ...order,
    userName: await fetchUser(order.userId),
  }))
).process(orders);
// All fetchUser() calls run concurrently — order is preserved

Immutable builder pattern

const base = pipe(filter<Product>((p) => p.inStock));

// base is not mutated
const names  = base.pipe(map<Product, string>((p) => p.name));
const prices = base.pipe(map<Product, number>((p) => p.priceCents / 100));

const [nameList, priceList] = await Promise.all([
  names.process(products),
  prices.process(products),
]);

ApiClient integration

import { ApiClient, pipe, filter, map } from "bytekit";

const client = new ApiClient({ baseUrl: "https://api.example.com" });

const products = await client.get<RawProduct[]>("/products", {
  pipeline: pipe(
    filter<RawProduct>((p) => p.active),
    map<RawProduct, Product>((p) => ({
      id: p.id,
      name: p.name,
      price: p.price_cents / 100,
    }))
  ),
});
// Pipeline applied after response parsing/validation

Error handling

try {
  await pipe(
    map<number, number>((n) => {
      if (n < 0) throw new RangeError("negative");
      return Math.sqrt(n);
    })
  ).process([4, 9, -1]);
} catch (err) {
  // err is RangeError("negative") — original error, not wrapped
}