reduce

What it does

In noy-db, records are encrypted before they reach any storage backend, and queries run inside the hub, after decryption — the store only ever sees ciphertext. The reduce service adds summary computation to that query layer: count, sum, avg, min, max reducers and groupBy bucketing on query(), computed in-memory over the records that match your where/filter clauses. For collections too large for the in-memory path, the streaming scan().aggregate() terminal reduces record-by-record with memory bound by the number of reducers, not the number of records.

Typical uses: per-period totals (sum('amount') grouped by month), per-segment averages, status counts, and multi-reducer rollups in one pass. The service is a tree-shake seam — when you don't opt in, its ~886 lines of Reduction/GroupedQuery machinery never reach your bundle.

Prerequisites

• None — withReduce() alone enables .aggregate() and .groupBy() on every collection query.

Opt in

import { createNoydb } from '@noy-db/hub'
import { withReduce } from '@noy-db/hub/reduce'
import { toMemory } from '@noy-db/to-memory'

const db = await createNoydb({
  store: toMemory(),
  user: 'alice',
  secret: 'a-strong-secret',
  reduceStrategy: withReduce(),
})

Usage

The flows below share this setup — a vault with a few invoices:

import { count, sum, avg } from '@noy-db/hub/reduce'

type Invoice = { clientId: string; period: string; status: string; amount: number }

const vault = await db.openVault('acme')
const invoices = vault.collection<Invoice>('invoices')

await invoices.put('inv-1', { clientId: 'c1', period: '2026-05', status: 'paid', amount: 100 })
await invoices.put('inv-2', { clientId: 'c1', period: '2026-06', status: 'paid', amount: 250 })
await invoices.put('inv-3', { clientId: 'c2', period: '2026-05', status: 'paid', amount: 850 })
await invoices.put('inv-4', { clientId: 'c2', period: '2026-05', status: 'open', amount: 5000 })

One-shot totals

query().aggregate(spec) takes a record of named reducers and returns an Reduction; .run() executes synchronously (the eager query path reads the already-decrypted cache) and returns one result field per reducer:

const totals = invoices.query()
  .where('status', '==', 'paid')
  .aggregate({ total: sum('amount'), n: count(), mean: avg('amount') })
  .run()
// → { total: 1200, n: 3, mean: 400 }

Reduction always runs over the full match set — .limit() / .offset() are ignored (a paginated aggregation would be a different operation).

Grouped totals

.groupBy(field) between the clauses and .aggregate(spec) partitions the matches into buckets and reduces each bucket. Every result row carries the group key plus the reducer outputs:

const byClient = invoices.query()
  .where('status', '==', 'paid')
  .groupBy('clientId')
  .aggregate({ total: sum('amount'), n: count() })
  .run()

[
  { "clientId": "c1", "total": 350, "n": 2 },
  { "clientId": "c2", "total": 850, "n": 1 }
]

Multi-key groupBy

groupBy(...fields) also accepts multiple fields; rows then carry every grouped field in declaration order:

const byClientAndPeriod = invoices.query()
  .where('status', '==', 'paid')
  .groupBy('clientId', 'period')
  .aggregate({ total: sum('amount') })
  .run()

[
  { "clientId": "c1", "period": "2026-05", "total": 100 },
  { "clientId": "c1", "period": "2026-06", "total": 250 },
  { "clientId": "c2", "period": "2026-05", "total": 850 }
]

Bucket identity is order-insensitive — .groupBy('a', 'b') and .groupBy('b', 'a') produce the same buckets (the internal canonical key sorts field names); the declaration order only determines the order of the key fields on each output row. Buckets are emitted in first-seen order — .sort() the rows downstream if you need a specific ordering.

Streaming aggregation over large collections

scan().aggregate(spec) is an async terminal that reduces the scan stream one record at a time. Memory is O(reducers), never O(records), which makes it the right tool for collections that don't fit in memory:

const paid = await invoices.scan()
  .where('status', '==', 'paid')
  .aggregate({ total: sum('amount'), n: count() })
// → { total: 1200, n: 3 }

Each scan().aggregate() call pays a full scan; for several aggregates in one pass, put them all in one spec.

API

All value exports below are available from @noy-db/hub/reduce (preferred for new code) and re-exported from the @noy-db/hub root for backward compatibility.

withReduce()

withReduce(): ReduceStrategy

Builds the strategy that activates .aggregate() / .groupBy() on queries. Pass it as reduceStrategy to createNoydb(...).

Reducer factories

Every factory accepts a ReducerOptions bag:

Exact money aggregation: collections that declare a moneyFields map get it automatically — at .aggregate() time, sum/min/max over a declared money field are rewritten to exact BigInt reducers that finalize to a decimal string (never a float). Union-form materialized views aggregate over a concatenated stream with no backing collection context, so they opt in via the moneyFields map on the withMaterializedView strategy instead.

query().aggregate(spec)

aggregate(spec: ReduceSpec): Reduction<ReduceResult<Spec>>
aggregate(build: (b: ReducerBuilder<T, S, M>) => Spec): Reduction<ReduceResult<Spec>>

ReduceSpec is a plain record of named reducers; the result carries one field per spec key. The builder overload types field arguments against the collection's record type, refuses sensitive fields at compile time, and types sum/min/max over declared money fields as MoneyString:

const totals = invoices.query()
  .aggregate((b) => ({ total: b.sum('amount'), n: b.count() }))
  .run()
// → { total: 6200, n: 4 }

Terminals on Reduction<R>:

• .run(): R — execute synchronously.
• .live(): LiveAggregation<R> — reactive value that recomputes whenever the source collection changes.

query().groupBy(...fields)

groupBy<F>(field: F): GroupedQuery<T, F, S, M>          // single key
groupBy<F>(...fields: F): GroupedQueryN<T, F, S, M>     // two or more keys

The only operation on a grouped query is .aggregate(spec) (same bare-spec and builder overloads as above), which returns a GroupedReduction with three terminals:

• .run(): R[] — one row per bucket, group key fields first, reducer outputs after.
• .runAsync({ locale?, fallback? }): Promise<R[]> — like .run(), and when the single grouping field is a dictionary-key field (see i18n) and a locale is given, appends a resolved <field>Label to each row. Single-field groupings only.
• .live(): LiveAggregation<R[]> — reactive grouped rows.

Ordering, limiting, and filtering belong on the query before .groupBy(); there is no having / post-group ordering.

scan().aggregate(spec)

scan().aggregate(spec: ReduceSpec): Promise<ReduceResult<Spec>>
scan().aggregate(build: (b: ReducerBuilder<T, S, M>) => Spec): Promise<ReduceResult<Spec>>

Async, streaming, O(reducers) memory. Honors the scan's .where() / .filter() clauses. Works with or without withReduce() — the streaming reducer loop lives in the core scan builder. scan().groupBy() and scan().aggregate().live() do not exist.

LiveAggregation<R>

Returned by .live() on both terminals. Framework-agnostic reactive primitive:

// Await it: `live()` hydrates the collection first, and reading an
// un-awaited handle on an unloaded collection throws
// `CollectionNotHydratedError` — the aggregate path carries the same guard
// as a plain `query().live()` (verified against @noy-db/hub 0.8.0-pre.0).
const open = await invoices.query()
  .where('status', '==', 'open')
  .aggregate({ total: sum('amount') })
  .live()

open.value // → { total: 5000 }
open.subscribe(() => console.log(open.value)) // returns an unsubscribe fn
// … later
open.stop()

Reducer<R, S>

Reducers are pure data: init() builds the initial state, step(state, record) folds a record in, finalize(state) collapses to the user-visible result. Optional remove(state, record) un-folds a record (admits incremental maintenance) and merge(a, b) combines partial states (must be associative + commutative). Any object with this shape plugs into an .aggregate() spec:

import type { Reducer } from '@noy-db/hub/reduce'

const distinct = (field: string): Reducer<number, Set<unknown>> => ({
  init: () => new Set(),
  step: (s, r) => s.add((r as Record<string, unknown>)[field]),
  finalize: (s) => s.size,
})

invoices.query().aggregate({ clients: distinct('clientId') }).run()
// → { clients: 2 }

Constants and errors

• GROUPBY_WARN_CARDINALITY (10_000) — distinct-group count at which a one-shot console warning fires (deduplicated per field set).
• GROUPBY_MAX_CARDINALITY (100_000) — hard cap; the run throws GroupCardinalityError (exported from @noy-db/hub) the moment the cap would be exceeded.

Advanced exports

For building on the service directly: the Reduction, GroupedQuery, GroupedQueryN, GroupedReduction classes, the pure pipeline functions reduceRecords(records, spec) and groupAndReduce(records, fields, spec, moneyFields?), buildLiveReduction(recompute, upstreams), the shared reducerBuilder instance, and resetGroupByWarnings() (test helper that clears the cardinality-warning dedup). Types: ReduceStrategy, ReduceSpec, ReduceResult, Reducer, ReducerOptions, ReducerBuilder, GroupedRow, GroupedRowN, LiveAggregation, AggregationUpstream.

Behavior when NOT opted in

• query().aggregate(...) and query().groupBy(...) throw immediately: "Aggregate / groupBy is not enabled on this Noydb instance. Import { withReduce } from "@noy-db/hub/reduce" and pass it to createNoydb({ reduceStrategy: withReduce() })."
• scan().aggregate(...) still works — the streaming terminal is part of the core scan builder and doesn't use the heavy Reduction / GroupedQuery classes.
• None of the ~886 LOC reaches the bundle.

Limits & gotchas

• groupBy cardinality is capped: one-shot warning at 10,000 distinct groups, GroupCardinalityError at 100,000. The caps are fixed constants, not options. Grouping by a high-uniqueness field (id, createdAt) is almost always a query mistake; beyond the cap you want streaming scan().aggregate() per segment or an external OLAP system.
• Full match set only: .aggregate() ignores .limit() / .offset().
• Non-number coercion: sum/avg/min/max coerce non-number field values (null, undefined, strings, objects) to 0 rather than producing NaN. Filter upstream with .where() if you need strict behavior.
• Money typing: sum() over a declared money field returns an exact decimal string (or a per-currency map in multi-currency mode without convertTo) even though its static type says number — use moneySum/moneyMin/moneyMax for correctly-typed results. A money-shaped value ({ amount, currency }) reaching a plain numeric reducer on an undeclared field throws, telling you to declare it in the collection's moneyFields.
• Bucket keys: records missing the group field (undefined) and records with an explicit null get separate buckets. Coalesce upstream if you want them merged.
• .live() recomputes fully on every source change (no incremental per-delta maintenance yet) and must be released with stop().

See also

• Query basics — the where/filter/orderBy pipeline aggregation sits on
• indexing — the .where() clauses ahead of an aggregation dispatch through indexes
• live — reactive queries; LiveAggregation follows the same subscribe/stop contract
• i18n — dictionary-key fields whose labels .runAsync({ locale }) resolves per row
• materialized views — multi-key groupBy inside materialized-view queries, and UNION sources with strategy-level moneyFields
• Analytics app recipe
• Proven by showcase 10-with-aggregate.