@pyreon/rx is a library of signal-aware reactive data transforms. Every collection and aggregation function is overloaded: pass a Signal<T[]> and you get back a Computed that auto-tracks and re-derives whenever the source changes; pass a plain T[] and you get a one-shot static result. The same filter / map / sortBy / groupBy you reach for on arrays, but they participate in Pyreon's reactivity graph for free.

It is not a lodash / es-toolkit replacement. It is a reactive operator library — its value is that a derived view stays in sync with its source signal with zero plumbing. For pure, one-off array utilities on plain data, keep using es-toolkit.

Installation

npm install @pyreon/rx

bun add @pyreon/rx

pnpm add @pyreon/rx

yarn add @pyreon/rx

Peer dependency: @pyreon/reactivity.

Quick Start

import { rx } from '@pyreon/rx'
import { signal } from '@pyreon/reactivity'

interface User {
  name: string
  age: number
  department: string
  active: boolean
}

const users = signal<User[]>([
  { name: 'Alice', age: 30, department: 'eng', active: true },
  { name: 'Bob', age: 25, department: 'eng', active: false },
  { name: 'Charlie', age: 35, department: 'design', active: true },
])

// Signal in → Computed out (auto-reactive)
const active = rx.filter(users, (u) => u.active) // Computed<User[]>
const sorted = rx.sortBy(active, 'name') // Computed<User[]>
const top10 = rx.take(sorted, 10) // Computed<User[]>

// top10() re-derives automatically whenever `users` changes
top10() // [{ name: 'Alice', ... }, { name: 'Charlie', ... }]

Why Rx?

Without @pyreon/rx, deriving a filtered-and-sorted view of a signal means hand-writing a computed and remembering to call the source inside it:

// Manual — you write the computed, you call the source, you chain by hand
const view = computed(() => {
  const arr = users()
  return arr
    .filter((u) => u.active)
    .slice()
    .sort((a, b) => (a.name < b.name ? -1 : 1))
    .slice(0, 10)
})

With @pyreon/rx the wiring is the function. Each transform detects the signal, allocates the tracked computed, and reads the source for you:

// Reactive — the function IS the wiring
const view = rx.take(rx.sortBy(rx.filter(users, (u) => u.active), 'name'), 10)

Or, flattened with pipe into one computed (not three — see pipe):

const view = pipe(
  users,
  (u) => u.filter((x) => x.active),
  (u) => [...u].sort((a, b) => (a.name < b.name ? -1 : 1)),
  (u) => u.slice(0, 10),
)

Signal Overloading

Every collection/aggregation function inspects its source argument and branches:

• source is callable (a signal, computed, or any function) → it allocates a computed, calls the source inside it, and returns the Computed. The result auto-tracks and re-derives on every source change.

• source is a value (a plain array) → it runs once and returns the plain result. No signals involved.

// Reactive — returns Computed<User[]>
const a = rx.filter(usersSignal, (u) => u.active)
a() // auto-tracks; re-derives when usersSignal changes

// Static — returns User[]
const b = rx.filter([{ active: true }], (u) => u.active)
// just a plain filtered array, nothing reactive

Two import styles

@pyreon/rx ships both a namespaced rx object (dot-notation) and individual named exports (tree-shakeable). They are identical functions.

import { rx } from '@pyreon/rx'
const a = rx.filter(users, (u) => u.active)

import { filter, sortBy, take } from '@pyreon/rx'
const a = filter(users, (u) => u.active)

Collections

The 23 collection transforms. Each is overloaded Signal<T[]> → Computed / T[] → plain. The reactive return type is noted per function.

filter

Keep items matching a predicate. The predicate receives (item, index).

const active = rx.filter(users, (u) => u.active) // Computed<User[]>
const big = rx.filter([1, 2, 3, 4, 5], (n) => n > 3) // [4, 5] (plain)

map

Transform each item to a new value. The mapper receives (item, index).

const names = rx.map(users, (u) => u.name) // Computed<string[]>

flatMap

Map each item to an array and flatten one level — exactly Array.prototype.flatMap. Empty inner arrays drop out, so it doubles as a filter-and-map in one pass.

const allTags = rx.flatMap(posts, (p) => p.tags) // Computed<string[]>
rx.flatMap([1, 2, 3], (n) => [n, n * 10]) // [1, 10, 2, 20, 3, 30] (plain)
rx.flatMap([1, 2, 3], (n) => (n % 2 === 0 ? [n] : [])) // [2] — empties drop out

The mapper must return an array (n => [n * 2], not n => n * 2), and it flattens exactly one level. Prefer flatMap over a map(...) + flatten(...) pair — that's two computed nodes; flatMap is one.

sortBy

Sort by a key name or a key-selector function — KeyOf<T> = keyof T | ((item: T) => string | number) — with an optional direction ('asc' default, 'desc' to invert). It is non-mutating (copies via [...arr]) and uses a plain a < b ? -1 : a > b ? 1 : 0 comparison.

const byName = rx.sortBy(users, 'name') // by key name, ascending
const newest = rx.sortBy(users, 'createdAt', 'desc') // descending
const byAge = rx.sortBy(users, (u) => u.age) // by key-selector

groupBy

Group items into buckets by a key / key-selector. Returns a plain Record<string, T[]>, not a Map. Keys are String(...)-coerced; insertion order within each bucket is preserved.

const byDept = rx.groupBy(users, (u) => u.department) // Computed<Record<string, User[]>>
for (const [dept, members] of Object.entries(byDept())) {
  /* ... */
}

countBy

The counting companion to groupBy — buckets by key but returns per-bucket counts (Record<string, number>), single-pass. Keys are String(...)-coerced, same as groupBy.

const perRole = rx.countBy(users, 'role') // Computed<Record<string, number>> → { admin: 2, viewer: 1 }
rx.countBy([1, 2, 2, 3], (n) => (n % 2 === 0 ? 'even' : 'odd')) // { odd: 2, even: 2 }

Use countBy for counts and groupBy for the members — countBy avoids the mapValues(groupBy(...), (g) => g.length) two-node detour.

keyBy

Index items into a Record<string, T> by a key / key-selector. Last write wins on a key collision.

const byId = rx.keyBy(users, 'id') // Computed<Record<string, User>>
byId()['42'] // the user with id 42

uniqBy

Deduplicate by a key / key-selector — keeps the first occurrence of each key. For primitive values without a key, use unique.

const oneEach = rx.uniqBy(users, 'email') // Computed<User[]>

unique

Deduplicate primitive values via a Set (keeps first occurrence). For objects, use uniqBy with a key.

rx.unique([1, 2, 2, 3, 1]) // [1, 2, 3]
const distinctIds = rx.unique(idsSignal) // Computed<number[]>

take / skip / last

Slice from the front, drop from the front, or take from the end.

const first5 = rx.take(users, 5) // Computed<User[]> — items 0..4
const rest = rx.skip(users, 5) // Computed<User[]> — items 5..end
const last3 = rx.last(users, 3) // Computed<User[]> — last 3 items

first

The first element (or undefined if empty).

const head = rx.first(items) // Computed<T | undefined>

chunk

Split into fixed-size chunks. The final chunk holds the remainder.

const pages = rx.chunk(users, 10) // Computed<User[][]>
// [[...10], [...10], [...rest]]

flatten

Flatten one level of nesting (Array.prototype.flat()).

const flat = rx.flatten(nestedSignal) // Signal<T[][]> → Computed<T[]>

find

The first item matching a predicate (or undefined).

const admin = rx.find(users, (u) => u.role === 'admin') // Computed<User | undefined>

compact

Drop all falsy values (null, undefined, false, 0, '', NaN).

rx.compact([0, 1, null, 2, '', 3, false]) // [1, 2, 3]

reverse

Reverse the order — non-mutating (returns a new copy, unlike Array.prototype.reverse).

rx.reverse([1, 2, 3]) // [3, 2, 1]
const descByAge = rx.reverse(rx.sortBy(users, (u) => u.age)) // sortBy is ascending; reverse → descending

partition

Split into a [pass, fail] tuple by a predicate. The predicate receives (item, index).

const [even, odd] = rx.partition([1, 2, 3, 4], (n) => n % 2 === 0)
// even: [2, 4], odd: [1, 3]
// Signal input → Computed<[T[], T[]]>

takeWhile / dropWhile

Take from the start while the predicate holds (stops at the first failing item); or drop the leading run that matches and return the rest. Predicate receives (item, index).

rx.takeWhile([1, 2, 3, 1, 2], (n) => n < 3) // [1, 2]  — stops at the first 3
rx.dropWhile([1, 2, 3, 1, 2], (n) => n < 3) // [3, 1, 2] — drops leading 1,2

intersection / difference / union

Set operations by identity or by key selector — signal-aware on BOTH inputs (either may be a signal; the result recomputes when either changes). O(n + m) via a Set. union is order-preserving (source first, first occurrence wins).

const selectedIds = signal([2, 4])
const rows = signal<Row[]>([...])

const selectedRows = rx.intersection(rows, selectedList, 'id') // rows also in selectedList
const unselected = rx.difference(rows, selectedList, 'id')     // rows NOT in selectedList
const merged = rx.union(localRows, serverRows, 'id')           // dedup by id, local wins

sample

Pick n random items via a Fisher-Yates partial shuffle. Returns all items if n >= length.

rx.sample([1, 2, 3, 4, 5], 2) // e.g. [3, 1]

mapValues

Map over the values of a Record — the natural follow-up to groupBy. The mapper receives (value, key).

const byDept = rx.groupBy(users, (u) => u.department)
const counts = rx.mapValues(byDept, (members) => members.length)
// Computed<Record<string, number>> → { eng: 2, design: 1 }

Aggregation

Eight functions that collapse a collection to a single value. Signal input → Computed<scalar>; plain input → the scalar.

count

The item count.

const total = rx.count(users) // Computed<number>

sum

Sum numeric values. With no key it sums the items themselves; with a key / key-selector it sums that field. Values are coerced via Number(...).

rx.sum([1, 2, 3]) // 6
const totalAge = rx.sum(users, (u) => u.age) // Computed<number>
const totalAge2 = rx.sum(users, 'age') // by key name

average

Mean of numeric values (optional key / key-selector). Returns 0 for an empty collection, not NaN.

const avgAge = rx.average(users, 'age') // Computed<number>

min / max

Find the item with the smallest / largest numeric value (optional key / key-selector). Returns the item, not the value — and undefined for an empty collection.

const youngest = rx.min(users, (u) => u.age) // Computed<User | undefined> — the User, not the age
const oldest = rx.max(users, 'age') // Computed<User | undefined>
const oldestAge = rx.max(users, 'age')()?.age // read the value off the returned item

reduce

Fold to a single value with an explicit initial accumulator. The reducer receives (acc, item, index).

const total = rx.reduce(items, (acc, item) => acc + item.price, 0) // Computed<number>

every / some

Whether all / any items match a predicate. Predicate receives (item, index).

const allActive = rx.every(users, (u) => u.active) // Computed<boolean>
const hasAdmin = rx.some(users, (u) => u.role === 'admin') // Computed<boolean>

Operators

Five signal-to-signal operators. Unlike the collection transforms, several of these operate on values over time rather than array contents.

distinct

Skip consecutive duplicate emissions from a signal. Uses Object.is by default, or a custom equals. Returns a ReadableSignal<T> & { dispose } — like the timing operators, distinct owns an eager effect() that is auto-torn-down inside a component / effectScope; call .dispose() for standalone use.

const status = signal('idle')
const changes = rx.distinct(status) // only emits when the value actually changes
const byId = rx.distinct(item, (a, b) => a.id === b.id) // custom equality

scan

A running accumulator over a signal's changes — like reduce, but it emits the accumulated value on every source change. Returns a ReadableSignal<U> & { dispose }. The reducer runs on the current source value immediately (the initial source() is the first emission), and like distinct it is scope-aware — auto-torn-down in a component, .dispose() for standalone use.

const clicks = signal(0)
const total = rx.scan(clicks, (acc, val) => acc + val, 0)
// clicks → 1 ⇒ total 1 ; clicks → 3 ⇒ total 4 ; clicks → 2 ⇒ total 6

combine

Combine the values of 2 or 3 signals into one Computed. The signals are passed as separate arguments, with the combiner function last (typed for 2 or 3 sources).

const first = signal('Ada')
const last = signal('Lovelace')
const fullName = rx.combine(first, last, (f, l) => `${f} ${l}`) // Computed<string>
const label = rx.combine(name, age, dept, (n, a, d) => `${n} (${a}, ${d})`) // 3 sources

zip

Pair up multiple arrays element-by-element into tuples, truncating to the shortest. Reactive if any input is a signal (typed for 2 or 3 sources).

const names = signal(['Alice', 'Bob'])
const ages = signal([30, 25])
const pairs = rx.zip(names, ages) // Computed<[string, number][]> → [['Alice', 30], ['Bob', 25]]

merge

Concatenate multiple arrays into one (flattens one level). Reactive if any input is a signal.

const a = signal([1, 2])
const b = signal([3, 4])
const all = rx.merge(a, b) // Computed<number[]> → [1, 2, 3, 4]

Timing

Two operators that delay or rate-limit a signal's value (the whole emitted value, not individual array items). Both return a ReadableSignal<T> with an extra .dispose() method, and both seed synchronously with the current source().

debounce

Emit the latest value only after the source has been quiet for ms milliseconds. Ideal for search-as-you-type.

const raw = signal('')
const debounced = rx.debounce(raw, 300) // ReadableSignal<string> & { dispose }

effect(() => {
  fetchResults(debounced()) // fires 300ms after the user stops typing
})

// Inside a component this is auto-cleaned on unmount. Standalone, dispose:
onCleanup(() => debounced.dispose())

throttle

Emit at most once per ms — a steady max rate during continuous change (emits immediately on first change, then waits the interval). Ideal for scroll / resize / pointermove.

const throttled = rx.throttle(scrollY, 100)
effect(() => updateHeader(throttled()))
onCleanup(() => throttled.dispose())

Search

search(source, query, keys) is a case-insensitive substring filter across the named string fields. It is reactive when either source or query is a signal — making search-as-you-type a one-liner. An empty / whitespace query returns the full list.

const query = signal('')
const results = rx.search(users, query, ['name', 'email']) // Computed<User[]>
// query = "ali" → matches "Alice" (substring, case-insensitive)

Pipe

pipe(source, ...transforms) composes transforms left-to-right and collapses the entire chain into one computed (not one per step). Each transform is a plain function that receives the resolved value from the previous step and returns the next — (value) => newValue. A signal source yields a Computed; a plain-array source yields a one-shot plain result. Typed for up to 7 transforms, with full type narrowing across steps.

This is pipe's structural payoff over chaining separate rx calls. A 3-step chain built as const a = filter(src, …); const b = sortBy(a, …); const c = take(b, …) allocates 3 computed nodes and does 3 recomputes per source change (each node re-derives in turn); the same chain in a single pipe is 1 node, 1 recompute — measured exactly by bun run --filter=@pyreon/rx bench:

Prefer pipe for any chain longer than 2 steps — fewer nodes, fewer subscription links, ~1 computed retained (~913 B) instead of ~N×913 B.

const topRisks = pipe(
  findings,
  (items) => items.filter((f) => f.severity === 'critical'),
  (items) => [...items].sort((a, b) => b.score - a.score),
  (items) => items.slice(0, 10),
)
// topRisks() — reactive Computed, re-derives when `findings` changes

Transforms can change the type at each step — the final type flows through:

const summary = pipe(
  numbers, // Signal<number[]>
  (arr) => arr.filter((n) => n % 2 === 0), // number[]
  (arr) => arr.reduce((sum, n) => sum + n, 0), // number
  (total) => `Total: ${total}`, // string
)
summary() // "Total: 180"

TypeScript

The package exports its two supporting types:

import type { KeyOf, ReadableSignal } from '@pyreon/rx'

// KeyOf<T> = keyof T | ((item: T) => string | number)
//   — the key/key-selector accepted by sortBy, groupBy, keyBy, uniqBy, sum, min, max, average
// ReadableSignal<T> = (() => T) & { peek?: () => T }
//   — any callable that returns a value (a signal, computed, or accessor)

Reactive returns are typed as the Computed produced by @pyreon/reactivity's computed; static (plain-array) returns are typed as the bare result. The overloads resolve the right return type from your source argument.

API Reference

rx namespace

rx is a Readonly object exposing all 42 functions plus pipe for dot-notation use (rx.filter(...)). Every member is also a tree-shakeable named export.

Collections (23)

Aggregation (8)

Operators (5)

Timing (2)

Both seed synchronously with source(); both must be .dispose()-d to release their internal effect + timer.

Search (1)

Case-insensitive substring match across the named string fields; reactive if source or query is a signal; empty query returns the full list.

Pipe (1)

Composes left-to-right into one computed; signal source → Computed, plain source → one-shot value. Transforms are plain (value) => newValue functions (not curried rx operators).

Types