Aggregates

Not what you want? For an on-demand count / sum / groupBy / window that runs when called (not on a schedule), see Aggregations — the query-builder methods. This page is the SCHEDULED, materialised *.aggregate.ts convention.

Use an aggregate for a pre-defined query whose result should be computed once and served fast for many reads. Top-N leaderboards. Most-used X over Y. Daily/weekly summary snapshots. Anything where the source query is expensive but the result is small and bounded.

The framework discovers *.aggregate.ts files at boot, runs the build function on a schedule, holds the rows in a cache, and serves them via useAggregate(def).read(...). You don't write a backing table, a refresh timer, a staleness check, or a dashboard surface — they're all part of the convention.

#Why this instead of *.cron.tsx

You could put the same logic in a cron handler: query the source, delete the cache table, insert the new rows. The aggregate convention earns its slot by owning seven things cron leaves to you:

Capability	Cron handler	*.aggregate.ts
Auto-create + manage backing storage	❌ user defines table + migration	✅ framework owns lifecycle
Atomic refresh (no empty-window for readers)	❌ DELETE-then-INSERT shows empty briefly	✅ transactional swap
refreshedAt metadata per aggregate	❌ track manually	✅ first-class
Staleness-aware reads (maxAgeMs throw)	❌ none	✅ explicit
Refresh overlap handling	⚠️ generic cron	✅ aggregate-specific
meta.lastError for failures	❌ user logs	✅ surfaced
Dashboard integration	⚠️ generic cron entry	✅ aggregate-specific (rows, last error, staleness)

#File shape

// apps/api/aggregates/topPlayers.aggregate.ts
import { defineAggregate } from '@voltro/runtime'
import { Schema } from 'effect'
import { gt, gte, and } from '@voltro/database'
import { database } from '../database/index'
import { yearStart } from '../lib/time'

export const TopPlayer = Schema.Struct({
  playerId: Schema.String,
  rank:     Schema.Number,
  kd:       Schema.Number,
})
export type TopPlayer = Schema.Schema.Type<typeof TopPlayer>

export default defineAggregate({
  name: 'topPlayers',
  refresh: '1h',                     // simple interval string

  output: TopPlayer,                  // every row decoded against this

  // Optional B-tree indexes on the DB-backed storage — honoured on the
  // SQL stores; a no-op on the in-memory store (no secondary indexes).
  indexes: [['rank']],

  // The build function. Same `ctx.store` a mutation handler sees.
  build: async (ctx) => {
    const players = await ctx.store.query(
      database.players
        .where(and(gt('lastMatchAt', yearStart()), gte('kdRatio', 3.0)))
        .orderBy('kdRatio', 'desc')
        .take(100)
        .descriptor,
    )
    return players.map((p, i) => ({
      playerId: p.id,
      rank:     i + 1,
      kd:       p.kdRatio,
    }))
  },
})

The default export must be the return value of defineAggregate({...}). The cli identifies files by checking for that branded shape — extra exports (like TopPlayer for the row type) are fine and conventional.

#Reading from a handler

import { useAggregate } from '@voltro/runtime'
import topPlayersDef, { type TopPlayer } from '../aggregates/topPlayers.aggregate'

export default (input, _ctx) => Effect.gen(function* () {
  const handle = yield* useAggregate(topPlayersDef)
  const top10: ReadonlyArray<TopPlayer> = yield* handle.read({ limit: 10 })
  return { top10 }
})

The reading API is explicit + namespace-only — useAggregate(def).read(...), not database.topPlayers.findAll(). This is a deliberate API choice (see Why no virtual-table integration below).

#Read options

handle.read({
  where:    { teamId: 't1' }, // filter the materialised rows (see below)
  limit:    10,            // pagination
  offset:   20,
  orderBy:  'rank',        // any column in the output schema
  direction: 'desc',
  maxAgeMs: 10 * 60_000,   // throws AggregateStale if last refresh older
})

#Parameterised reads — where

Without a filter an aggregate can only ever be "the one global roll-up". Every tenant-, team- or period-scoped roll-up — which is most of the real ones — then has to read the whole aggregate and filter client-side: every row crosses the wire so the caller can throw most of them away. where moves that cut to the read.

// one team's rows, for one year
const rows = yield* handle.read({ where: { teamId: 'team_7', year: 2026 } })

// an array is an IN set — status is 'open' OR 'blocked'
const active = yield* handle.read({ where: { status: ['open', 'blocked'] } })

// composes with the other read options
const top = yield* handle.read({
  where:   { teamId: 'team_7', status: ['open', 'blocked'] },
  orderBy: 'rank',
  limit:   10,
})

The semantics, exactly:

• Entries are ANDed — a row matches only when it satisfies every entry.
• A scalar value means strict equality (===) against that field on the row.
• An array value means IN — the row's value must be one of the array's entries.
• An omitted where, or an empty {}, filters nothing.
• Filtering happens before orderBy and limit / offset, so pagination paginates the filtered set.

where is deliberately data, not a predicate function. It is applied over the rows the aggregate has already materialised — the refresh still computes the full roll-up, and where cuts the result before it crosses the wire. Keeping it a serializable record of field → value (rather than a callback) is what leaves the door open to pushing the same filter down to the store later. A cut you can't express as equality / IN belongs in another aggregate rather than in the read.

#Metadata

For dashboards, "data was last refreshed X minutes ago" UI labels, and operational health:

const meta = yield* handle.meta()
// {
//   refreshedAt: Date | null,           // null until first refresh
//   rowCount: number,                    // last refresh result count
//   durationMs: number,                  // how long the build function took
//   lastError: string | null,            // last refresh error (null on success)
//   nextRefreshAt: Date | null,          // when the next interval fires
// }

lastError keeps the last failure visible without dropping the previous good rows — failing refreshes preserve last-good data. Reading meta after a failed refresh tells you the run failed; reading the rows still serves the last successful result.

#Manual refresh

yield* handle.refresh()

Triggers an immediate refresh from the current handler. Use for "Run now" dashboard buttons or for cron-fed callers (your *.cron.tsx can drive aggregate refresh in custom windows that the aggregate's own interval policy doesn't cover).

#Refresh policies

Three forms — all three honoured at runtime:

refresh: '5m'                                                            // interval (string)

refresh: { cron: '0 2 * * *', timezone: 'UTC' }                         // cron expression

refresh: { interval: '1h', onChange: ['matches'], debounceMs: 30_000 }  // hybrid

• Interval string — fires every X. Parses <n><unit> where unit is ms / s / m / h / d. Invalid strings fail boot loudly.
• Cron expression — { cron: '<expr>', timezone: '<iana-tz>' }. The framework computes next-firing via the same compileCron / Cron.next the *.cron.tsx engine uses.
• Hybrid — { interval, onChange: [<table>, ...], debounceMs }. The interval is the upper bound; on every change to one of the listed tables, the framework debounces by debounceMs before firing. Use for "freshness on write but capped staleness".

#Incremental maintenance (IVM)

By default an aggregate recomputes — it re-runs its build query on each refresh (interval / cron / onChange). For a bounded aggregate you can opt into incremental view maintenance: each CDC delta on the source table updates the maintained value in place, with no full re-query.

export default defineAggregate({
  name: 'sales.byRegion',
  refresh: { onChange: ['orders'] },
  output: SalesByRegion,
  incremental: {
    source: 'orders',
    op: 'sum',                 // exactly one of count / sum / avg / min / max
    column: 'amount',          // required for everything but count
    groupBy: ['region'],       // one maintained row per group (optional)
    toRow: (g) => ({ region: g.group['region'] as string, total: g.value }),
  },
})

The shape must be maintainable — defineAggregate throws at import (via classifyShape) otherwise:

• exactly one op, from count / sum / avg / min / max;
• a single source table (no joins);
• no window functions, no distinct, no having.

How it maintains, per delta on the source:

• count / sum / avg — applied directly (insert adds, delete subtracts, update adjusts); O(1) per change.
• min / max — applied directly on insert/raise; a delete/lower of the current extreme triggers a bounded rescan of that one group to find the new extreme.

When a shape can't be incremental, leave incremental off and use a refresh policy — the engine recomputes. Incremental is an optimisation for hot, bounded aggregates, not a different result.

#source is a table name, and it is checked

incremental.source is typed against your app's own tables — the same TableName a query's source: uses — so a misspelled or renamed-away name is a compile error from your next voltro dev.

That check exists because the failure has no other symptom. A source matching no table does not error: the runner subscribes to something nothing writes, no delta ever arrives, and the aggregate quietly stops tracking its input while every read of it still succeeds and still returns a number.

The boot says it too. An aggregate whose CDC source resolves to no live table is named in the same stale-source: warning queries and streams appear in, on both boot paths — a name can be correct at the type level and still be a table this deployment does not have. An aggregate with no incremental block declares no source and is not audited.

#Boot strategy

Each aggregate declares bootRefresh (default 'persistent'). All three modes are wired in the aggregate runner (attachAggregates):

Mode	Behaviour	Use when
'persistent' (default)	Storage is rehydrated from the prior run's rows, so reads return the previous-good result immediately on restart; an async refresh then runs in the background and the first scheduled refresh fires at its natural time.	Most aggregates — best UX, no empty window on restart.
'async'	Backing storage is reset at boot; the first refresh fires on the next tick, so reads return [] until it completes.	Test/dev fresh-start, or when stale data on restart would mislead.
'sync'	Boot BLOCKS until the first refresh completes — slow boot, never serves stale data.	Pricing / business-critical computations.

A fresh-boot empty window only happens under 'async'. If that would cause real bugs and you don't want 'sync''s slow boot, gate reads on meta().refreshedAt (or use read({ maxAgeMs }) to throw AggregateStale).

#Failure semantics

A build function that throws keeps the last-good rows in the cache. The error message lands in meta.lastError. Reads continue to serve the prior result.

This is intentional: a failing refresh shouldn't take down reads of the previous good aggregate. Last-good-data beats empty-on-fail.

A build function that returns rows that fail output decode also records lastError (output schema mismatch: row[N] failed output-schema decode at <path>: <issue>). The cache stays at the last successful refresh.

A second refresh starting while the first is still running is skipped — overlap-handling is 'skip' by design. Long-running build functions don't pile up calls; they just slow the effective refresh rate.

#Storage backends

The framework picks per-store:

• Memory store → in-process state, per-replica. Lost on restart; multi-instance fans out duplicate work. Fine for dev / single-pod deployments.
• SQL stores (postgres / mysql / mariadb / mssql / sqlite / turso) → cross-dialect JSON-payload backing (_voltro_aggregate_meta + _voltro_aggregate_rows tables, auto-created on first boot). Atomic refresh via a generation counter — readers always see either the old generation or the new, never a mid-swap mix. Per-process in-memory mirror keeps reads sub-millisecond on the hot path.

#Cluster coordination

Aggregates inherit the framework's existing coordination story (plan 45). On store: 'memory' the runner uses singleCoordinator (one process); on real SQL stores it uses an advisoryLockCoordinator so multi-instance deployments fire exactly-one refresh per scheduled tick. No config needed — the runner picks based on the resolved store dialect.

#Inspect surface

Two endpoints under /_voltro/inspect/:

GET  /_voltro/inspect/aggregates                    — snapshot all + meta
POST /_voltro/inspect/aggregates/<name>/refresh     — manual "Run now"

The cloud dashboard's Aggregates panel consumes these. Same auth resolver gates the surface as the rest of /_voltro/inspect/* (VOLTRO_INSPECT_TOKEN by default).

#Refresh strategy — 'replace' vs 'merge'

strategy controls how each refresh applies its rows:

defineAggregate({
  name: 'topPlayers',
  refresh: '1h',
  output: TopPlayer,
  strategy: 'replace',   // default — every refresh fully replaces the previous result
  build: async (ctx) => { ... },
})

defineAggregate({
  name: 'playerWins',
  refresh: '5m',
  output: PlayerWins,
  strategy: 'merge',
  mergeKey: 'playerId',  // required for 'merge' — each row's identity field
  build: async (ctx) => {
    // Only return players whose stats CHANGED since the last refresh.
    // Players not returned this time KEEP their previous values.
    return getPlayersWithRecentActivity()
  },
})

'replace' (default) — every refresh fully replaces the previous result. Rows from a prior refresh that the new build didn't return are dropped. Best for top-N leaderboards, summaries, snapshot-style aggregates.

'merge' — upsert by mergeKey. Each row from build() either UPDATEs an existing entry (matched by row[mergeKey]) or INSERTs a new one. Prior rows whose key isn't in the new batch STAY. Best for "track per entity" patterns where each refresh only needs to recompute the changed entities (active players in the last hour, weapons used today, etc.). The mergeKey field MUST exist on every row your build returns — defineAggregate throws at registration if strategy: 'merge' is set without a mergeKey.

Mental model: 'replace' is "snapshot at time T"; 'merge' is "incremental delta into a long-lived aggregate". Pick 'replace' when an aggregate's value is the FULL recomputation; pick 'merge' when each refresh contributes only the deltas.

#Why no virtual-table integration

The defining property of an aggregate is the query is fixed in advance. Treating it as a query-buildable virtual table (database.topPlayers.where(...)) opens four footguns:

1. Hidden staleness. database.topPlayers.where(...) looks like a live query. Readers can't tell it's stale data.
2. Computation drift. A full query builder shifts arbitrary computation from refresh-time to read-time — the materialisation point IS the query; don't re-query it. read({ where }) is the bounded exception: an equality / IN cut of rows that are already materialised, not a new query.
3. Misleading expectations. Users would reflexively try database.topPlayers.insert(...). Framework would either silently do nothing or error with a cryptic message.
4. Cross-timeline joins. Joining an aggregate with a live table mixes two timelines (refresh-time + now). Mostly a footgun.

The explicit namespace (useAggregate(def).read(...)) makes the materialisation explicit. where covers the one cut that genuinely belongs at read time — scoping a roll-up to a tenant, a team, a period. Everything past it (joins, aggregating over the aggregate, arbitrary predicates) keeps its friction on purpose: it pushes you to either define another aggregate or do the work in app code with clear boundaries.

#Decision: aggregate vs subscriber vs cron

What you want	Use
In-transaction work tied to a write	mutation handler
Reject the insert	table().validate(Schema) (schema DSL)
Derive a value at write time	column.computed(row => ...) / column.default(() => ...)
Best-effort post-commit per-row reactivity	*.subscribe.ts
Crash-safe post-commit reactivity	workflow invoked from *.subscribe.ts
Periodically-refreshed pre-computed query	*.aggregate.ts (this)
Event ingestion + time-bucketed aggregates over events	Analytics sink
Real OLAP / 50M+ rows / arbitrary SQL	warehouse plugin (ClickHouse, DuckDB, Tinybird)

#Cross-plan: querying the warehouse from a build function

The build function gets ctx.analytics alongside ctx.store — the configured AnalyticsSink. ctx.store reads the main DataStore (OLTP); ctx.analytics reads the warehouse (OLAP: DuckDB / ClickHouse / postgres-lite). This is the first-class path for "reduce 10B warehouse events to a 100-row leaderboard materialised in the main store for fast reads":

import { defineAggregate } from '@voltro/runtime'
import { Effect, Schema } from 'effect'

export const TopPlayer = Schema.Struct({
  playerId: Schema.String,
  rank:     Schema.Number,
  wins:     Schema.Number,
})

export default defineAggregate({
  name: 'topPlayers',
  refresh: '1h',
  output: TopPlayer,
  indexes: [['rank']],
  build: (ctx) =>
    Effect.gen(function* () {
      // Warehouse query — reduces millions of events to the top 100.
      const top = yield* ctx.analytics.topN({
        event:   'match_completed',
        groupBy: 'playerId',
        metric:  'count',
        n:       100,
        range:   { from: new Date(Date.now() - 30 * 86_400_000) },
      })
      return top.map((entry, i) => ({ playerId: entry.key, rank: i + 1, wins: entry.value }))
    }),
})

ctx.analytics exposes the full sink contract — topN, aggregate, timeseries, and track. Like ctx.store, the build function runs as the system (no per-request subject).

When no analytics sink is configured the framework provides the no-op sink: the read methods fail with AnalyticsCapabilityNotSupported({ provider: 'noop' }), which surfaces as a refresh error (last-good rows are preserved). Provider-specific queries beyond the four-method contract (raw SQL, HyperLogLog) live outside the sink — there is no raw-client escape hatch; query the provider with your own client instance inside the build function where you need them.

#Standing IVM siblings — expectations, cost budgets, experiments

Three more primitives are built on the same engine as defineAggregate({ incremental }) — they reduce a table to a value maintained incrementally from store.onChange CDC deltas (O(1) per write, no re-query). Unlike an aggregate you never read() a materialised table; you observe a live signal through a framework registry and a useX(def) handler-sugar. Each is discovered by its own file convention and wired into both voltro dev and voltro serve.

Primitive	File	Reduces a table to	Observe with
defineExpectation	*.expectation.ts	a holding/violated data-quality signal	useExpectation / ExpectationRegistry
defineCostBudget	*.budget.ts	per-tenant compute-cost attribution + ok/warn/exceeded budgets	useCostBudget / CostRegistry
defineExperiment	*.experiment.ts	per-variant metric + lift (live A/B / holdout)	useExperiment / ExperimentRegistry

They are soft, observability-grade signals — none of them ever blocks a write. Rejecting a write is a business rule's job (table().validate(Schema)); running an agent/workflow on a change is *.reaction.tsx. These three only watch and report.

#Data-quality expectations (*.expectation.ts)

A defineExpectation is a standing data-quality contract over a table. Where dbt tests / Great Expectations run in BATCH and catch bad data hours later, an expectation re-evaluates on every write and tips holding↔violated the instant an invariant breaks — tracing the violation to the write that caused it (the CDC event's traceId / subjectId / procedure).

// apps/api/expectations/orderFreshness.expectation.ts
import { defineExpectation } from '@voltro/runtime'

export default defineExpectation({
  name: 'orders-fresh',
  on: { table: 'orders' },      // its CDC deltas drive re-evaluation
  invariant: { kind: 'freshness', column: 'createdAt', maxAgeMs: 10 * 60_000 },
  severity: 'critical',         // 'info' | 'warn' (default) | 'critical' — alerting priority only
})

The default export must be the return value of defineExpectation({...}) — the cli discovers *.expectation.ts files by that branded shape.

Four invariant kinds, each reducing the table to one incrementally-maintained metric:

{ kind: 'freshness',   column: 'createdAt', maxAgeMs: 600_000 }              // newest row no older than X (re-checked on a clock tick too)
{ kind: 'nullRate',    column: 'email', maxRate: 0.01 }                      // ≤ 1% of rows null in `column`
{ kind: 'rowCount',    min: 1, max: 100_000 }                               // COUNT(*) within [min, max]
{ kind: 'valueBounds', column: 'price', min: 0, max: 1000, maxViolationRate: 0.05 } // ≤ 5% of rows out of [min, max]

An optional on.where predicate narrows the population the invariant is asserted over — it runs server-side per row and never leaves the server, so it can be any predicate:

on: { table: 'orders', where: (row) => (row['status'] as string) === 'paid' }

Observe one expectation from a handler with useExpectation, or read the whole surface off ExpectationRegistry:

import { Effect } from 'effect'
import { useExpectation } from '@voltro/runtime'
import ordersFresh from '../expectations/orderFreshness.expectation'

export default (input, _ctx) =>
  Effect.gen(function* () {
    const state = yield* useExpectation(ordersFresh)
    // state: { status: 'holding' | 'violated' | 'unknown', metric, threshold,
    //          since, lastEvaluatedAt, lastCause, ... } | null (null = not registered)
    return { degraded: state?.status === 'violated' }
  })

ExpectationRegistry exposes snapshot() (every expectation's state — the inspect feed), get(name), violations() (the alerting view), and subscribe(listener) for violated/recovered transitions. A transition carries the causing write's provenance; a freshness SLA aging out with no write reports cause: null — the honest "no write caused this; the ABSENCE of writes did".

#Cost budgets (*.budget.ts)

A defineCostBudget is a reactive-FinOps primitive: per-tenant / per-subscription compute-cost attribution plus a budget every tenant is held to independently. It is to reactive compute what AI's requireAiBudget is to USD spend — a standing per-tenant ceiling that crosses ok→warn→exceeded and recovers on a window rollover.

// apps/api/budgets/tenantCompute.budget.ts
import { defineCostBudget } from '@voltro/runtime'

export default defineCostBudget({
  name: 'tenant-compute',
  limit: 100_000,       // the per-tenant ceiling, in the budget's unit
  unit: 'recompute',    // which cost unit to meter; omit ⇒ the tenant's TOTAL across every unit
  warnAt: 0.8,          // fraction of `limit` at which it goes 'warn' (default 0.8; set 1 to disable)
  window: '24h',        // tumbling window — the counter resets each boundary; omit ⇒ cumulative since boot
  severity: 'warn',     // 'info' | 'warn' (default) | 'critical'
})

Observe one budget for one tenant with useCostBudget(def, tenantId):

import { Effect } from 'effect'
import { useCostBudget } from '@voltro/runtime'
import tenantCompute from '../budgets/tenantCompute.budget'

export default (input, ctx) =>
  Effect.gen(function* () {
    const state = yield* useCostBudget(tenantCompute, ctx.tenantId)
    // state: { status: 'ok' | 'warn' | 'exceeded', spent, limit, warnThreshold, ... } | null
    return { overBudget: state?.status === 'exceeded' }
  })

CostRegistry exposes attribution() (per-tenant chargeback/showback rows — total + byUnit + bySubscription), tenant(tenantId), budgets(), budget(name, tenantId), breaches() (the alerting view), and subscribe(listener) for threshold crossings.

Reactive recomputes now feed attribution automatically. Declaring any *.budget.ts wires the dispatcher's recordCost tap in both boot paths: every time a source-row change re-runs an affected subscription and pushes it a delta, one { unit: 'recompute', amount: 1 } cost event is attributed to that subscription's tenant (and traceId). So a tenant's total/spent populates from real reactive work — a unit: 'recompute' budget crosses ok→warn→exceeded as deliveries accrue, and recovers on a window rollover. The emission is per delivered recompute — one event per subscription the change fanned out to, on both the row-set and computed-query delivery paths. An app that declares NO cost budget wires no tap and allocates nothing on the reactive hot path.

#Online experiments (*.experiment.ts)

A defineExperiment is a live A/B / holdout experiment expressed as IVM aggregates. Assignment is a deterministic salted hash (subject → variant, no stored assignment table, reproducible on the client); the success metric is maintained PER VARIANT from the watched table's CDC — so lift vs a baseline is real-time, with no batch pipeline. It differs from plugin-flags (which GATES a code path) — an experiment MEASURES the outcome.

// apps/api/experiments/checkoutColor.experiment.ts
import { defineExperiment } from '@voltro/runtime'

export default defineExperiment({
  name: 'checkout-button-color',
  on: { table: 'orders' },                      // its CDC deltas drive the live recompute
  subject: 'userId',                            // stable per-subject bucketing (a column name, or a (row) => key fn)
  variants: [{ name: 'control' }, { name: 'green', weight: 1 }],   // ≥ 2 arms; `weight` skews the split (default 1)
  holdout: 0.1,                                 // 10% held out entirely, for a clean untouched baseline
  metric: { kind: 'conversionRate', column: 'completed' },         // per-variant success metric
  baseline: 'control',                          // which variant lift is measured against (default: the first)
})

Metric kinds: { kind: 'count' }, { kind: 'sum', column }, { kind: 'avg', column }, and { kind: 'conversionRate', column, equals? } (converted iff row[column] is truthy, or === equals when given). avg / conversionRate are per-subject rates that a lift reads honestly; count / sum reflect exposure too, so their cross-variant comparison only means "more/less total".

Observe the live result with useExperiment, or off ExperimentRegistry:

import { Effect } from 'effect'
import { useExperiment } from '@voltro/runtime'
import checkoutColor from '../experiments/checkoutColor.experiment'

export default (input, _ctx) =>
  Effect.gen(function* () {
    const result = yield* useExperiment(checkoutColor)
    // result: { totalExposure, baseline, variants: [{ variant, isBaseline, isHoldout,
    //           exposure, metric, lift, diff }, ...], lastUpdatedAt, ... } | null
    return { arms: result?.variants ?? [] }
  })

Each variant row carries its exposure (sample size), the maintained metric, and — for non-baseline arms — lift ((metric − baseMetric) / baseMetric) and diff (absolute). ExperimentRegistry exposes snapshot(), get(name), and subscribe(listener) for the live recompute stream that a results view redraws from. The same-subject-same-variant assignment is a pure function (assignVariant(def, subject)) a client can reproduce.