pipelined (@pplnd/core)

Typed errors, explicit optional values, and composable pipelines for TypeScript.

Note: pipelined is pre-1.0. The API may change between minor versions until the 1.0 release.

npm add @pplnd/core

Overview

An API call fails, and the error arrives as a caught unknown three functions away from the call that caused it. Every return type along the way says the call succeeded.

pipelined puts that failure back into the signature. Result carries a typed error or a value, Maybe carries presence or absence, and Task.Result carries both across an async boundary. The containers are plain objects with a kind field, functions take their data last, and multi-step work reads top to bottom in a pipe.

It has zero dependencies, ships ESM and CommonJS builds, and tree-shakes down to ~27 B–6.57 kB gzipped for individual modules and 26.79 kB for the entire package.

Documentation

Full guides and API reference at pplnd.dev.

If you are deciding between modules, start with Choosing a Container. To see several modules joined in one feature, read the recipes. Every module name in the quick reference links to its guide.

---

Examples

Four modules below show the shape: a typed failure, an async lifecycle, a request policy, and a data pipeline. Every other module follows the same conventions; the quick reference links each guide, and every guide carries a complete example of its own.

Handling errors without throw (Result)

Functions that throw exceptions leave error handling to caller memory. TypeScript does not track thrown errors in return types, and catch (err) types errors as unknown. Result<E, A> makes failure a normal return value:

import { pipe, Result } from "@pplnd/core";

type Config = { apiKey: string; port: number };
type ConfigError =
  | { kind: "InvalidJson"; message: string }
  | { kind: "MissingField"; field: string };

const parseConfig = (raw: string): Result<ConfigError, Config> =>
  pipe(
    Result.tryCatch(
      (): unknown => JSON.parse(raw),
      { onError: (err): ConfigError => ({ kind: "InvalidJson", message: String(err) }) },
    ),
    Result.ensure(
      (data): data is { apiKey: string; port?: unknown } =>
        typeof data === "object" && data !== null && "apiKey" in data
        && typeof data.apiKey === "string",
      (): ConfigError => ({ kind: "MissingField", field: "apiKey" }),
    ),
    Result.map((data) => ({
      apiKey: data.apiKey,
      port: typeof data.port === "number" ? data.port : 8080,
    })),
  );

const outcome = parseConfig('{"apiKey": "secret-123"}');

if (Result.is.ok(outcome)) {
  console.log(`Starting server on port ${outcome.value.port}`);
} else {
  console.error(`Config error: ${outcome.error.kind}`);
}

The compiler enforces that outcome.value is only accessible after checking Result.is.ok(outcome). If parsing or validation fails, the pipeline returns a typed error.

Lazy async workflows and cancellation (Task.Result)

Standard promises start executing immediately upon creation and reject with unknown. Task.Result<E, A> is a lazy function (signal?: AbortSignal) => Deferred<Result<E, A>>. Its Deferred can be awaited. Nothing runs until you invoke the task, errors remain typed, and chained tasks receive the signal. Each task must pass that signal to work that supports cancellation:

import { pipe, Result, Task } from "@pplnd/core";

type User = { id: string; name: string };
type Post = { id: string; title: string };
type ApiError = { status: number; message: string };

const fetchUser = (id: string): Task.Result<ApiError, User> =>
  Task.Result.tryCatch(
    async (signal) => {
      const res = await fetch(`/users/${id}`, { signal });
      if (!res.ok) throw new Error(res.statusText);
      return (await res.json()) as User;
    },
    { onError: (err): ApiError => ({ status: 500, message: String(err) }) },
  );

const fetchPosts = (userId: string): Task.Result<ApiError, Post[]> =>
  Task.Result.tryCatch(
    async (signal) => {
      const res = await fetch(`/users/${userId}/posts`, { signal });
      if (!res.ok) throw new Error(res.statusText);
      return (await res.json()) as Post[];
    },
    { onError: (err): ApiError => ({ status: 500, message: String(err) }) },
  );

// Chain two requests — the AbortSignal propagates to both steps automatically
const userWithPosts = (id: string) =>
  pipe(
    fetchUser(id),
    Task.Result.chain((user) =>
      pipe(
        fetchPosts(user.id),
        Task.Result.map((posts) => ({ ...user, posts })),
      )
    ),
  );

To run the pipeline with cancellation support:

const controller = new AbortController();
const run = userWithPosts("42");

// Execution starts here; aborting controller cancels the active fetch request
const outcome = await run(controller.signal);

if (Result.is.ok(outcome)) {
  console.log(outcome.value); // { id: "42", name: "...", posts: [...] }
} else {
  console.error(outcome.error.message); // ApiError, not unknown
}

Managing request lifecycles, retries, and concurrency (Op)

UI events often arrive faster than the network can settle: a search box fires on every keystroke, a user double-clicks a submit button, or a flaky connection needs exponential backoff. Writing this by hand requires managing AbortController instances, timers, and attempt counters.

Op separates what a request does from how overlapping calls and failures are handled:

import { Duration, Op, RetryPolicy, Task } from "@pplnd/core";

type SearchResult = { id: string; title: string };
type SearchError = { kind: "NetworkError"; cause: unknown } | { kind: "Timeout" };

const searchOp = Op.make((query: string) =>
  Task.Result.tryCatch(
    (signal) =>
      fetch(`/search?q=${encodeURIComponent(query)}`, { signal }).then((r) =>
        r.json() as Promise<SearchResult[]>
      ),
    { onError: (cause): SearchError => ({ kind: "NetworkError", cause }) },
  )
);

const search = Op.interpret(searchOp, {
  strategy: "restartable", // newer runs replace older ones and signal them to abort
  retry: RetryPolicy.exponential({
    attempts: 3,
    initial: Duration.milliseconds(200),
  }),
  timeout: {
    duration: Duration.seconds(5),
    onTimeout: (): SearchError => ({ kind: "Timeout" }),
  },
});

search.subscribe((state) => {
  if (Op.is.pending(state)) console.log("Searching...");
  if (Op.Outcome.is.ok(state)) console.log(state.value);
  if (Op.Outcome.is.err(state)) console.error(state.error.kind);
});

const input = document.querySelector<HTMLInputElement>("#search");
input?.addEventListener("input", () => search.run(input.value));

For form submissions, swapping strategy: "restartable" to strategy: "exclusive" automatically drops duplicate clicks while a request is in flight. Built-in strategies include restartable, exclusive, debounced, throttled, queue, buffered, concurrent, keyed, and once.

Transforming data collections (Arr & Rec)

Chaining native array methods like .map(), .filter(), and .sort() allocates intermediate arrays at each step, and index lookups return undefined without type guidance. Arr and Rec provide curried helpers designed for pipe:

import { Arr, Maybe, Num, pipe, Rec, Str } from "@pplnd/core";

type RawItem = { name: string; price: string; category: string };
type Item = { name: string; price: number; category: string };

const normalise = (raw: RawItem): Maybe<Item> =>
  pipe(
    Num.from.string(raw.price), // "9.99" → Some(9.99), "n/a" → None
    Maybe.map((price) => ({
      name: Str.trim(raw.name),
      price,
      category: raw.category,
    })),
  );

const cheapestByCategory = (items: RawItem[]) =>
  pipe(
    items,
    Arr.filterMap(normalise), // parse + drop unparseable rows in a single pass
    Arr.sortBy((a, b) => a.price - b.price), // sort ascending by price
    Arr.groupBy((item) => item.category), // Record<string, Arr.NonEmpty<Item>>
    Rec.map((group) => Arr.NonEmpty.head(group)), // lowest price per category without Maybe wrapper
  );

Arr.filterMap combines mapping and filtering into a single loop. Arr.groupBy returns guaranteed non-empty groups, allowing Arr.NonEmpty.head to extract the first element directly without an unnecessary Maybe wrapper.

---

Quick Reference

Each section links a problem to the module that solves it. To choose between neighbouring modules first, read Choosing a Container; for complete features built from several modules, read the recipes.

Error Handling and Optionality

Problem to Solve	Module
Synchronous typed errors without try/catch or throw	Result
Optional values without null / undefined guards	Maybe
Multi-field form and batch validation error accumulation	Validation
Eliminating impossible UI loading/error/data states	RemoteData
Values that can hold one side, the other, or both at once	These
Pure calculation audit trails and decision logging	Logged

Async, Concurrency, and State

Problem to Solve	Module
Lazy async workflows with automatic AbortSignal propagation	Task.Result, Task
Managing request race conditions, retries, timeouts, and locks	Op
In-memory typed event dispatching, sequence funnels & queue safety	EventBus
Setup and reverse-order cleanup of connections and locks	Resource
Type-safe finite state machines & validated transitions	Machine
Declarative DAG task scheduling & cycle detection	Graph
Infallible async synchronization latch for tasks	Deferred
Circuit breaker pattern for protecting failing upstream services	CircuitBreaker
Multi-step workflows with optional reverse-order compensation	Transaction
Request batching and deduplication within a batch	Batcher
Keyed reuse of recent successful lookups with expiry and eviction	Cache
Recognising a cancelled run across async engines	Aborted
Pure state transitions, tokenizers, and parsers	State
Dependency injection and contextual configuration without drilling	Reader

Collections and Optics

Problem to Solve	Module
Bounds-safe array transformations & single-pass filters	Arr, Arr.NonEmpty
Type-safe object manipulation & key migration	Rec, Rec.NonEmpty
Immutable Map operations with any key type	Dict, Dict.NonEmpty
Immutable Set operations (union, intersection, diff)	Uniq, Uniq.NonEmpty
Deep nested immutable updates without spread boilerplate	Lens, Optional
String sanitization, numeric conversion & slug parsing	Str, Str.NonEmpty
Boundary clamping & division-by-zero protection	Num
Exact bigint math without floating-point rounding errors	BigNum
Boolean checks and conditional branching in pipelines	Bool
Safe JSON parsing & circular reference protection	Json

Composition and Utilities

Problem to Solve	Module
Left-to-right value pipeline execution	pipe
Asynchronous value & function pipeline execution	pipeAsync, flowAsync
Left-to-right function composition	flow
Distinct compile-time types for primitive IDs	Brand
Explicit, unit-safe time spans & timeout policies	Duration, RetryPolicy
Deep value equality & custom comparison rules	Equality
Multi-column sorting with fallback tiebreakers	Ordering
Combining boolean checks and filter rules	Predicate
Runtime type narrowing & custom type guard composition	Refinement
Merging configs, counters, and summary objects	Combinable
Two-element [first, second] tuples	Pair
Conditional branching and object shaping in pipelines	when, unless, struct
Currying, uncurrying, tuple wrapping, and argument flipping	curry, uncurry, flip
Feeding one input into multiple functions and combining results	converge, juxt, on
Function result caching & pipeline side-effects	memoize, tap, tapLog
Small helper functions	identity, constant, noop, tuple

---

Package Architecture and Performance

The package declares "sideEffects": false, allowing supported bundlers to remove unused exports. Actual bundle sizes depend on the bundler, its settings, and the imports your application uses.

Standalone Module Footprint (Tree-Shaken)

These measurements include each listed import and its dependencies, minified and gzipped:

Import Pattern	Minified + Gzipped
pipe	142 B
flow	173 B
Brand	27 B
Duration	150 B
Maybe	1.51 kB
Result	1.51 kB
Validation	1.51 kB
RemoteData	1.94 kB
Machine	2.41 kB
EventBus	1.61 kB
Batcher	3.07 kB
Cache	2.40 kB
Resource	2.05 kB
Graph	2.88 kB
Task	4.55 kB
Transaction	2.55 kB
Op	6.30 kB
Lens	233 B
Arr	6.57 kB
Rec	3.25 kB
Dict	2.44 kB
Str	2.30 kB

See Performance and benchmarks for the benchmark suite and results.

Contributing

Contributions of every size are welcome. Start with CONTRIBUTING.md or the contributor handbook. It covers setup, the local commands, testing, module authoring, and guide writing.

Participation follows the Code of Conduct. Report security issues through SECURITY.md, not public issues.

License

BSD-3-Clause