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.

#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.