# Data

> How Voltro's reactive data layer works — queries, mutations, actions, streams, all over one WebSocket with typed errors and tracked dependencies.



---

<!-- source: en/data/overview.md -->
## Overview

_How Voltro's reactive data layer works — queries, mutations, actions, streams, all over one WebSocket with typed errors and tracked dependencies._

The api side of a Voltro app is a collection of typed RPC primitives. You drop descriptor/server-executor pairs into the api tree; the framework discovers them; the typed client lights up.

This section covers how data flows between server and client.

A form bound to a mutation, a table bound to a query, one reactive backend — add
a row and it appears instantly, pushed from the server (your own throwaway
sandbox; it resets on refresh):

```tsx
<AutoForm api="app" mutation="todos.create" submitLabel="Add todo" />
<DataTable api="app" query="todos.list" />
```

## The shape of it

```text
┌──────────────────────────────────────────────────────────────────┐
│  Client (React)                                                  │
│    useSubscription('app', 'notes.list', input)                   │
│    useMutation('app', 'notes.create')                            │
│    useAction('app', 'invites.send')                              │
│    useAgentStream('app', 'support.run')                          │
└──────────────────────────┬───────────────────────────────────────┘
                           │  @effect/rpc over WebSocket
                           ▼
┌────────────────────────────────────────────────────────────────────┐
│  api app                                                           │
│    *.query.ts    + *.query.server.ts     — reactive reads          │
│    *.mutation.ts + *.mutation.server.ts  — transactional writes    │
│    *.action.ts   + *.action.server.ts    — external I/O / unary    │
│    *.stream.ts   + *.stream.server.ts    — one-shot element push   │
│    *.route.tsx                            — public raw-HTTP route  │
│    *.workflow.tsx  + *.trigger.tsx        — durable work + events  │
│    *.agent.tsx                            — persisted AI chat      │
└────────────────────────────────────────────────────────────────────┘
```

Most of it goes through one WebSocket (REST routes are the public raw-HTTP exception). Queries and streams are both streaming RPCs, but they mean different things: a query emits reactive snapshot/delta envelopes; a stream emits plain elements and then finishes.

## Every primitive at a glance

Voltro ships more building blocks than this one section holds — each is a typed descriptor in a file the CLI discovers. The five at the top live here; the rest have their own sections but are the same kind of thing. The complete set:

| Primitive | File | What it is |
|---|---|---|
| **Query** | `*.query.ts` | reactive read → [Queries](/docs/data/queries) |
| **Mutation** | `*.mutation.ts` | transactional write → [Mutations](/docs/data/mutations) |
| **Action** | `*.action.ts` | unary external I/O → [Actions](/docs/data/actions) |
| **Stream** | `*.stream.ts` | one-shot element push → [Streams](/docs/data/streams) |
| **Event** | `*.event.ts` | ephemeral fan-out to many clients → [Events](/docs/data/events) |
| **REST route** | `*.route.tsx` | public raw-HTTP endpoint → [REST routes](/docs/data/rest-routes) |
| **Aggregate** | `*.aggregate.ts` | scheduled materialised query → [Aggregates](/docs/data/aggregates) |
| **Subscriber** | `*.subscribe.ts` | per-table post-commit reaction → [Subscribers](/docs/data/subscribers) |
| **Workflow** | `*.workflow.tsx` | durable multi-step work → [Workflows](/docs/workflows/overview) |
| **Event trigger** | `*.trigger.tsx` | event → workflow fan-out → [Event triggers](/docs/workflows/event-triggers) |
| **Schedule** | `*.cron.tsx` | cron-fired handler → [Scheduling](/docs/scheduling/overview) |
| **Agent** | `*.agent.tsx` | persisted AI chat → [Agents](/docs/ai/agents) |
| **Tool** | `*.tool.tsx` | model-callable function → [Tools](/docs/ai/tools) |
| **Seed** | `*.seed.ts` | boot/data seeding → [Seeds](/docs/database/seeds) |
| **Startup** | `*.startup.tsx` | run-once boot hook (long-lived work + teardown) → [Startup hooks](/docs/reference/startup) |
| **Email** | `*.email.tsx` | React-Email template → [Mail](/docs/plugins/mail) |
| **Webhook** | `*.webhook.tsx` | signed inbound / outbound → [Webhooks](/docs/plugins/webhooks) |

## What's in this section

**Primitives** — [Queries](/docs/data/queries) (reactive reads + dependency tracking) · [Mutations](/docs/data/mutations) (transactional writes, typed errors, auto-optimistic) · [Actions](/docs/data/actions) (unary RPC, no transaction) · [Subscriptions](/docs/data/subscriptions) (the reactive engine behind snapshots/deltas) · [Streams](/docs/data/streams) (`defineStream` element push) · [Events](/docs/data/events) (`defineEvent` — things that HAPPEN, with no row behind them) · [REST routes](/docs/data/rest-routes) (public raw-HTTP for third parties) · [Aggregates](/docs/data/aggregates) (scheduled materialised queries) · [Subscribers](/docs/data/subscribers) (per-table post-commit reactions).

**Protocol & errors** — [Wire protocol](/docs/data/wire-protocol) (framing, multiplexing) · [Error handling](/docs/data/errors) (Schema-tagged errors, retries, client narrowing).

## The contract

Every query, mutation, action, and stream has two files:

```text
queries/notes.list.query.ts          # descriptor: browser-safe schema + metadata
queries/notes.list.query.server.ts   # executor: server-only implementation
```

The descriptor defines the wire surface:

```ts
defineX({
  name:   'feature.action',
  input:  Schema.Struct({ /* ... */ }),
  output: Schema.Struct({ /* mutations/actions */ }),
  element: Schema.Struct({ /* streams */ }),
  error:  Schema.Union(ErrorVariantA, ErrorVariantB),
})
```

The server executor receives `(input, ctx)` and can return a plain value, `Promise`, `Effect`, query descriptor, or `Stream`, depending on the primitive. Everything is Schema-validated at the boundary. The TypeScript types flow to the client automatically through `rpcGroup.generated.ts`.

> **`name` charset.** A procedure `name` is `.`-separated and becomes a JS
> identifier in `rpcGroup.generated.ts` (`todos.list` → `todosListRpc`). It
> must start with a letter and use only letters, digits, and `.` / `-` / `_`.
> `-` and `_` are camelCased away, so two names differing only by separator
> **collide**. Codegen rejects an invalid or colliding name at `voltro dev`
> boot, naming the file. Prefer plain camelCase segments (`todos.list`).

## Reactivity model

Queries are live. A query descriptor declares `source: 'tableName'`; the server executor returns either a query builder/descriptor or a computed value. When a mutation writes to the table, the runtime recomputes affected subscriptions and pushes a snapshot or delta.

Streams are not live subscriptions. A stream executor returns a one-shot `Stream` of elements. Use streams for transient token feeds, progress events, logs, or external event sources. If the result should survive reloads or update other tabs, persist rows and expose them through a query.

## Tenant scoping

Every executor receives `ctx.request.subject` — the typed identity of the caller. Tables with the `tenant()` mixin auto-scope reads and writes. See [Multi-tenancy](/docs/multi-tenancy/overview) for the full story.

## When to use what

| You want… | Use |
|---|---|
| Read data that updates live | `*.query.ts` + `*.query.server.ts` + `useSubscription` |
| Write data atomically | `*.mutation.ts` + `*.mutation.server.ts` + `useMutation` |
| External I/O / unary side effect | `*.action.ts` + `*.action.server.ts` + `useAction` |
| Public HTTP endpoint for a third party (URL + JSON) | [`*.route.tsx`](/docs/data/rest-routes) (`defineRestRoute`) |
| Transient server-to-client element stream | `*.stream.ts` + `*.stream.server.ts` + `useAgentStream` |
| Durable persisted AI chat | `*.agent.tsx` or action + query over `agent_messages` |
| Background job | `*.workflow.tsx` |
| Fan a domain event out to one or more workflows | [`*.trigger.tsx`](/docs/workflows/event-triggers) + `ctx.events.publish(...)` |
| Pre-computed query result (top-N, summary) | [`*.aggregate.ts`](/docs/data/aggregates) |
| React to every commit on a table (server-side) | [`*.subscribe.ts`](/docs/data/subscribers) |
| Event ingestion + analytical aggregates over events | [Analytics sink](/docs/plugins/analytics) |

Pages don't have to pick only one. Most use `useSubscription` for reads and `useMutation` for writes; add actions or streams only when the workflow calls for them.



---

<!-- source: en/data/queries.md -->
## Queries

_`*.query.ts` + `*.query.server.ts` pairs — reactive reads consumed with useSubscription, with dependency tracking and delta updates._

A **query** is a reactive read. The client subscribes to it; when the underlying data changes, Voltro pushes a fresh snapshot or delta over the WebSocket. No polling, no manual `refetch`, no mutation response gymnastics.

Live — a computed-return query: the `{ open, done, total }` counts recompute the
instant you add or toggle a todo (it declares `source: 'todos'`):

```tsx
const stats = useSubscription('app', 'todos.stats')   // computed, live — no refetch
```

## Minimal query pair

Descriptor file:

```ts
// apps/api/queries/notes.list.query.ts
import { defineQuery } from '@voltro/protocol'
import { Schema } from 'effect'

export const listNotes = defineQuery({
  name:   'notes.list',
  source: 'notes',
  guards: [{ scope: 'notes:read' }],
  input:  Schema.Struct({}),
  output: Schema.Array(Schema.Struct({
    id:    Schema.String,
    title: Schema.String,
  })),
})
```

Server executor:

```ts
// apps/api/queries/notes.list.query.server.ts
import { database } from '../database/index'

export default () =>
  database.notes.orderBy('createdAt', 'desc').limit(100)
```

Save both files and `notes.list` becomes a streaming query in the typed client.

> **`guards:` is not decoration here — it is what makes the file boot.** Every
> wire-exposed procedure must declare exactly one of `guards:`,
> `openAccess: '<reason>'` or `internal: true`; a descriptor that declares none
> is refused at boot, naming the file. Which one is right is a real decision,
> and both of the other two appear on this page below. Full rules:
> [Authorization](/docs/authentication/authorization#every-procedure-decides-guards-or-openaccess).

## Consuming a query

```tsx no-check
import { useSubscription } from '@voltro/client'

export default function Notes() {
  const { data, loading, error } = useSubscription('app', 'notes.list', {})
  if (error) return <p>Error: {String(error)}</p>
  if (loading) return <p>Loading...</p>
  return (
    <ul>
      {data.map((note) => <li key={note.id}>{note.title}</li>)}
    </ul>
  )
}
```

The first argument is the api name from `app.config.ts.apis`; the second is the descriptor's `name`.

## Inputs

Descriptor:

```ts
// apps/api/queries/messages.list.query.ts
import { defineQuery } from '@voltro/protocol'
import { Schema } from 'effect'

export const listMessages = defineQuery({
  name:   'messages.list',
  source: 'messages',
  guards: [{ scope: 'messages:read' }],
  input:  Schema.Struct({
    channelId: Schema.String,
    limit:     Schema.Number,
  }),
  output: Schema.Array(Schema.Struct({
    id:        Schema.String,
    channelId: Schema.String,
    body:      Schema.String,
  })),
})
```

Server executor:

```ts
// apps/api/queries/messages.list.query.server.ts
import { eq } from '@voltro/database'
import { database } from '../database/index'

export default (input: { channelId: string; limit: number }) =>
  database.messages
    .where(eq('channelId', input.channelId))
    .orderBy('createdAt', 'desc')
    .limit(input.limit)
```

Client:

```tsx
const { data } = useSubscription('app', 'messages.list', {
  channelId: 'c_123',
  limit: 50,
})
```

### An undeclared field is refused, not dropped

The input schema is the whole accepted key set. A field it does not declare
fails the call with a `ParseError` naming the key and listing what was expected:

```
{ readonly channelId: string; readonly limit: number }
└─ ["employeeId"]
   └─ is unexpected, expected: "channelId" | "limit"
```

It used to be discarded silently, and the reason that is worse than it sounds is
what a discarded FILTER means. A deployment's layout called a `userSettings.list`
that declares `userId` with `{ employeeId }`; the payload decoded to `{}`, which
for a list query is not a narrower filter but the absence of one, and an admin
was served another user's row.

The decoder cannot tell a projection field from a filter field, so it refuses
either way. Three consequences worth knowing:

- Nested objects and union members follow the same rule — a stray key inside
  `{ page: { limit, offset } }` is refused too.
- `Schema.Struct({})` means "this procedure takes nothing", and a call carrying
  anything is refused. `Schema.Record(...)` keeps its open key set: there the
  openness is declared.
- If a call site legitimately holds more than the procedure declares — a spread
  of a wider filter object — narrow it at the call site rather than widening the
  schema. Widening restores the silent drop under a different name: the field is
  accepted and still does nothing.

## Descriptor-return vs computed-return

A query executor can return either:

| Executor returns | Use when |
|---|---|
| A `database.<table>` query builder / descriptor | You are streaming rows from one table and want fine-grained predicate-aware invalidation. |
| A computed value | You are building an aggregate, join, projection, or other derived shape. |

Computed example:

```ts
// apps/api/queries/notes.summary.query.ts
import { defineQuery } from '@voltro/protocol'
import { Schema } from 'effect'

export const notesSummary = defineQuery({
  name:   'notes.summary',
  source: 'notes',
  guards: [{ scope: 'notes:read' }],
  input:  Schema.Struct({}),
  output: Schema.Struct({
    open: Schema.Number,
    done: Schema.Number,
  }),
})
```

```ts
// apps/api/queries/notes.summary.query.server.ts
export default async (_input: Record<string, never>, ctx) => {
  const notes = await ctx.store.select('notes').all()
  return {
    open: notes.filter((note) => !note.done).length,
    done: notes.filter((note) => note.done).length,
  }
}
```

For computed queries, `source` is the reactivity trigger. When any row in the source table changes, the runtime re-runs the executor and emits the new value if it changed. If the executor reads more than one table (a join or matrix), declare `source` as an array — the executor re-runs when **any** listed table changes (e.g. `source: ['skills', 'ratings']`).

### Type the computed return against `output` — `defineExecutor`

A computed executor's return is encoded through the descriptor's `output` schema — but the bare default export's return is **not** type-checked against it. So a handler that builds `{ publishedAt: row.publishedAt.getTime() }` where `output` is `timestampMs` (Type `Date`) compiles green and throws `Expected DateFromSelf, actual 1784…` at *encode* time, which **Dies the subscription**. For a nullable date it's a time-bomb: fine while the value is null, exploding the instant it becomes non-null (a publish, say).

Wrap the handler in `defineExecutor(descriptor, fn)` — it pins the return to `Schema.Type<output>`, turning that into a compile error at the handler:

```ts
// apps/api/queries/notes.summary.query.server.ts
import { defineExecutor } from '@voltro/runtime'
import { notesSummary } from './notes.summary.query'

export default defineExecutor(notesSummary, async (_input, ctx) => {
  const notes = await ctx.store.select('notes').all()
  return {
    open: notes.filter((note) => !note.done).length,
    done: notes.filter((note) => note.done).length,
  }
})
```

It's a runtime identity (returns the handler unchanged) — the whole value is the compile check. The Effect error and requirement channels stay inferred; only the success value is constrained. A **descriptor-return** (reactive) executor is allowed through unchecked: the store produces the rows, so a value-level return type can't express the row-vs-`output` check. `defineExecutor` works the same for `defineMutation` / `defineAction` handlers.

> **Composing one executor inside another (e.g. a workflow `step`).** The value `defineExecutor` returns is typed as the `ExecutorReturn` UNION (`Output | Promise | Effect | descriptor-return`), so it has no `.pipe` — you can't feed the wrapped default export straight into another Effect. Export the handler's raw `Effect` separately (a named `export const execute = …`, or import the un-wrapped function) and compose THAT; keep the `defineExecutor`-wrapped default only as the procedure's entry point. The union return is deliberate — it's what lets one helper type every executor shape — so this is a "import the raw effect for composition" convention, not a gap to route around.

## Auto-optimistic source

`source` also connects query caches to mutation `target` metadata:

```ts
defineQuery({ name: 'notes.list', source: 'notes', guards: [{ scope: 'notes:read' }], /* ... */ })
defineMutation({
  name: 'notes.create', target: { table: 'notes', op: 'insert' },
  guards: [{ scope: 'notes:write' }], /* ... */
})
```

With that pairing, `useMutation('app', 'notes.create')` can stage an optimistic row in active `notes.list` caches without client-side cache plumbing.

### A guard that reads a second table belongs in `source`

`source` is the reactive trigger set: the query re-runs when a listed table
changes, and only then. So a guard that loads a row from ANOTHER table to decide
access has made that table part of what the result depends on:

```ts
export const teamBoard = defineQuery({
  name: 'boards.forTeam',
  input: Schema.Struct({ teamId: Schema.String }),
  output: BoardRows,
  // The guard resolves the caller's membership of THIS team — a read of
  // `teamMembers` that happens before the executor runs.
  guards: [{ action: 'view', resourceType: 'team', resource: (input) => input.teamId }],
  // `boards` alone is wrong here — that membership read is part of what the
  // result depends on.
  source: ['boards', 'teamMembers'],
})
```

Leave `teamMembers` out and the subscription does not re-run when membership
changes. **That is an authorization staleness, not a cosmetic one:** revoke
someone's membership and their open subscription keeps serving rows they may no
longer see, until something else happens to invalidate it.

Reported by a team whose own invariant caught it after five computed queries
under-declared their `source`; the fix was array sources.

### `voltro dev` now says when a query reads a table it did not declare

That paragraph used to end "nothing warns about this at runtime". It does now.

While `voltro dev` is running, every read a query makes is attributed to it and
compared against its own `source:`. Read a table you did not declare and the
terminal says so, once:

```
source: tasks.list: read `task_sub_tasks` without declaring it in `source:`.
  A write to that table will not re-run this query, so an open view keeps
  showing what it showed before. The write itself is fine, which is why nothing
  else reports this.
```

It is deliberately narrow, and knowing where its edges are is the difference
between acting on it and learning to skim it:

- **It reports what it SAW.** A branch that did not run contributes nothing, so
  it never claims your `source:` is otherwise complete — only that a table it
  watched you read is missing from it.
- **Once per query per boot.** A per-request warning on a hot list would be its
  own outage.
- **A query with no `source:` at all is left alone.** It has made no claim; the
  finding is about an incomplete list, not a missing one.
- **An eager-loaded relation COUNTS**, and it is the case worth knowing about.
  `.with({ subTasks: true })` issues no second read — the whole spec folds into
  one round trip — so the loaded table never appears as a read of its own. The
  recorder resolves it through the relation registry instead, target and (for a
  many-to-many) junction alike. A write to the junction changes membership,
  which is exactly the change a user makes.
- **A table read only to NARROW a result is not counted** — a parent reached
  through `inSubquery(...)`, or a read the framework made to resolve your row
  filter. Those decide which rows come back rather than contributing rows, and
  putting every one of them in `source:` would re-run every list on every
  membership write.

That last rule is a judgement the recorder makes for the common case and
deliberately does not make for yours. The section above is the case where you
want a restricting read in `source:` anyway — an authorization read whose
staleness you care about. The recorder will not nag you into it and will not
argue when you add it.

**Dev only.** `voltro serve` installs none of it — it costs a wrapper per read,
and a production log is not where this gets read. `VOLTRO_SOURCE_RECORDER=off`
turns it off in dev.

If your own helper resolves access somewhere the framework does not call it, wrap
it in `restrictingReads` (from `@voltro/runtime`) and its reads stop counting —
inside or outside a recording session, so it is safe to leave in place.

## `output` is the serializer — `timestampMs`

A descriptor's `output` is not documentation of the shape. It **is** the
serializer: it is handed to the rpc as the success schema, so a handler's result
is *encoded through it* on the way out and decoded on the client. Any conversion
the schema describes, the framework performs — you never need a converter at the
tail of a handler.

That is worth stating plainly, because the symptom is usually read backwards.
A `timestamp()` column comes back from the store as a `Date`, and `Date` is not
JSON. Declaring that field as `Schema.Number` looks like the fix, but it
describes the **wire** type rather than the domain type — which leaves the schema
with nothing to convert, and pushes the conversion back into the handler:

```ts no-check
// The shape that leads to hand-written converters everywhere
output: Schema.Array(Schema.Struct({ createdAt: Schema.Number })),
// …and then, at the tail of every executor:
return rows.map((row) => ({ ...row, createdAt: row.createdAt.getTime() }))
```

Declaring `Schema.DateFromNumber` instead makes the conversion automatic.
`@voltro/database/wire` ships that mapping under names you can drop straight
into your own struct — including the nullable case, which is the one that goes
wrong silently:

```ts
// apps/api/queries/projects.list.query.ts
import { defineQuery } from '@voltro/protocol'
import { timestampMs, timestampMsOrNull } from '@voltro/database/wire'
import { Schema } from 'effect'

export const listProjects = defineQuery({
  name:   'projects.list',
  source: 'projects',
  guards: [{ scope: 'projects:read' }],
  input:  Schema.Struct({}),
  output: Schema.Array(
    Schema.Struct({
      id:             Schema.String,
      name:           Schema.String,
      jiraProjectKey: Schema.String,
      addedAt:        timestampMs,        // Date in the handler, epoch ms on the wire
      archivedAt:     timestampMsOrNull,  // for a nullable timestamp column
      seenAt:         Schema.optional(timestampMs),
    }),
  ),
})
```

The executor returns its rows and computes nothing at the tail:

```ts no-check
// apps/api/queries/projects.list.query.server.ts
export default () => database.projects.orderBy('addedAt', 'desc').limit(100)
```

- `timestampMs` — `Date` in the handler, `number` (epoch ms) on the wire.
- `timestampMsOrNull` — for a `.nullable()` timestamp. Use this rather than
  converting a null by hand: `new Date(null)` is `1970-01-01`, so "never
  archived" renders as a plausible date instead of as nothing.
- An **optional** field is `Schema.optional(timestampMs)` — there is no third
  export for it.

This also matches the shape real handlers have. Most return a struct assembled
by hand across several tables — `{ id, name, slug, addedAt, jiraProjectKey }` —
where there is no single table to derive an output schema from anyway, and those
are exactly the places the hand-written `Date → epoch` converters pile up.

### Why the field schemas import from `@voltro/database/wire`

`@voltro/database/wire` is a **browser-safe** entry: it contains plain
`effect/Schema` values and imports `effect` and nothing else. That matters
because a descriptor is loaded **value-level by the web client** — the rpc client
needs every procedure's schema — so everything a `*.query.ts` transitively
imports ends up in the browser bundle.

The package root is not browser-safe, and neither is anything that reaches a
**table value**. Importing `../database/schema` to get at `notes` imports
`@voltro/database` itself, which pulls the store, the query builder and the SQL
driver into the client graph. `voltro dev` refuses to boot in that case and
prints the import chain:

```
browser-safety violation — the generated rpcGroup pulls a SERVER-ONLY module
into the client bundle.
  import chain:
      → rpcGroup.generated.ts
      → ./queries/notes.list.query
      → ../database/schema
      → @voltro/database
```

So a descriptor's `output` is always written as your own `Schema.Struct` with
field schemas — never derived from a table. Deriving from a table is a
server-side operation; see below.

### The Encoded / Type split

That split is the whole point — the handler works in domain types, the wire
carries something JSON can represent:

| Column | In the handler (Type) | On the wire (Encoded) |
|---|---|---|
| `timestamp()`, `date()` | `Date` | `number` (epoch ms) |
| `bigint()` | `bigint` | `string` (decimal) |
| `text()`, `enum()`, `id()`, `reference()` | `string` | `string` |
| `integer()`, `real()`, `decimal()` | `number` | `number` |
| `boolean()` | `boolean` | `boolean` |
| `json()`, `vector()`, `raw()` | `unknown` | `unknown` |

`bigint()` crosses as a **decimal string** on purpose: a `bigint` sent as a JSON
number rounds silently past 2^53, and a value that is quietly wrong is worse than
one that is rejected.

A `.nullable()` column wraps its mapped type, so a null timestamp round-trips as
`null` rather than becoming epoch 0 (which would render as a plausible
`1970-01-01` instead of "never").

### Row codecs for server-side code — `rowSchema`, `columnSchema`

`rowSchema(table)` builds the whole struct from a table definition, and
`columnSchema(def)` maps one column. Both take the table as a **value**, so by
the section above they can only be used in code that already runs on the server
alone — a `*.query.server.ts` / `*.mutation.server.ts` executor, a `*.seed.ts`,
a startup or job module, a maintenance script, a test. **Not** a descriptor's
`output`.

What they are for is encoding or decoding table rows outside the rpc path, where
no `output` schema is doing it for you: writing rows to a file export or a queue
payload, or validating seed / import data against the actual table shape before
it is written.

```ts no-check
// apps/api/database/notes.seed.ts — server-only, so the table value is fine
import { rowSchema } from '@voltro/database'
import { Schema } from 'effect'
import { notes } from './schema'

const rows = Schema.decodeUnknownSync(Schema.Array(rowSchema(notes)))(
  JSON.parse(await readFile('seed/notes.json', 'utf8')),
)
```

`omit` drops columns from the schema — the way to keep an internal column out of
an export:

```ts no-check
rowSchema(users, { omit: ['passwordHash'] })
```

It is a **convenience, not a security boundary**. The column is simply absent
from *this* schema; code that serializes the same row under a different schema
still emits it. For a real boundary, see
[`.encrypted()` and column sensitivity](/docs/database/sensitivity).

`columnSchema` takes a column **definition** — what `table.fields` holds — not
the builder that `text()` or `timestamp()` returns (a builder's definition is
private, so it cannot be read from outside):

```ts no-check
import { columnSchema } from '@voltro/database'

columnSchema(notes.fields.createdAt)   // ✅ a definition, from table.fields
columnSchema(timestamp())              // ❌ a builder — not readable
```

The mapping is single-sourced: `columnSchema` returns the very same
`timestampMs` value for a `timestamp()` column, so a derived row schema and a
hand-written descriptor struct can never disagree about the wire shape.

## What gets sent on the wire

Queries are streaming RPCs whose elements are **subscription events**: an initial `snapshot` followed by `delta`s. See [Wire protocol](/docs/data/wire-protocol#subscription-events-snapshot-delta) for the envelope shape. For plain element streams, use [Streams](/docs/data/streams).

## Anti-patterns

- **Putting server-only imports in `*.query.ts`.** Descriptors are imported by browser-safe codegen. Put database/SDK/filesystem imports in `*.query.server.ts`.
- **Mutating from a query executor.** Queries are reads. Use a mutation for writes.
- **Using a stream for durable data.** Streams are transient. Persist rows and expose them through a query when the UI should survive reloads or sync across tabs.


## Contradictions refused at declaration

```ts
defineQuery({ name: 'q', guards: [], … })              // ✗ enforces nothing
defineQuery({ name: 'q', source: '', … })              // ✗ reactive, subscribed to nothing
defineQuery({ name: 'q', internal: true, overridesPlugin: true, … })  // ✗ removes, replaces nothing
defineQuery({ name: 'q', openAccess: '', … })          // ✗ a marker with no reason
defineQuery({ name: 'q', guards: [{ scope: 'x' }], openAccess: 'open', … })  // ✗ two decisions
defineQuery({ name: 'q', internal: true, openAccess: 'open', … })  // ✗ no wire to decide about
```

The first three are the shapes `defineEvent` refuses too, for the same reasons — a rule that
holds for one primitive and not another is worse than no rule, because the
answer then depends on which file you happened to open.

**`guards: []`** reads at the call site as if the procedure were protected and
enforces nothing; the check runs only for a non-empty list. Omit the field.

**An empty `source`** declares reactivity and subscribes to nothing: the query
serves one snapshot and never updates, which is indistinguishable from "nothing
changed". Worse than a *stale* source, which the boot warning can at least name
— this one names no table at all, so nothing can report it.

**`internal: true` + `overridesPlugin`** removes the plugin's route and puts
something unreachable in its place: callers get a 404 for something that used to
work, with no diff that says so. Joins the existing refusals of `internal` with
`publicApi` or `exposeAsTool`.

**`openAccess` without a reason** is a marker that says nothing. The reason is
what a reviewer reads to decide whether this really should be callable without a
check — `openAccess: 'public pricing, no caller data'`.

**`openAccess` + `guards`** is two different access decisions at once: the
procedure is protected AND open. Keep the guards if a caller must hold a scope;
drop them if anyone may call it.

**`openAccess` + `internal: true`** decides about a surface that does not exist —
`internal` takes the procedure off the wire. Drop one of the two.

> Every wire-exposed procedure must carry ONE of `guards:` / `openAccess:` /
> `internal: true`, or the boot refuses it. See
> [Authorization](/docs/authentication/authorization) for the gate and the
> `security.defaultDeny` field that governs it.


## Loading vs empty — don't conflate them

`useSubscription` returns `loading` and `isEmpty` alongside `data`. They are
**different** states, and branching on `data === undefined` alone is what causes
a flash of empty-state before the first snapshot:

| State | Meaning | Render |
|---|---|---|
| `loading` | no snapshot has arrived yet | skeleton |
| `isEmpty` | snapshot arrived, zero rows (or a null value) | empty state |
| neither | rows present | the list |

### `loading` narrows `data`

`SubscriptionState<T>` is a **discriminated union on `loading`**, so `loading` is
not a flag sitting beside `data` — it is a type guard for it:

```ts
type SubscriptionState<T> =
  | { loading: true;  data: undefined; isEmpty: false }
  | { loading: false; data: T;         isEmpty: boolean }
// both members also carry revision, emittedAt, error and pendingPatches
```

Once you have returned for `loading`, `data` is `T`. No `?? []`, no `!`:

```tsx
const { data, loading, isEmpty } = useSubscription<Note[]>('app', 'notes.list', {})
if (loading) return <TableSkeleton/>
if (isEmpty) return <EmptyNotes/>
return <NotesTable notes={data}/>
```

A derived constant narrows just as well, as long as `loading` is part of it:

```tsx
const { data, loading } = useSubscription<Note[]>('app', 'notes.list', {})
const isLoading = !currentUser || loading
if (isLoading) return <TableSkeleton/>
return <NotesTable notes={data}/>   // data is Note[]
```

`fallback` fills `data` while loading so a page can render its real (empty) shell
immediately — it never lies about `loading`:

```ts
const { data, loading } = useSubscription('app', 'notes.list', {}, { fallback: [] })
// data is [] before the first snapshot; loading is still true
```

That call returns `SubscriptionStateWithFallback<T>` instead of the union: `data`
is always present (the fallback stands in until the first snapshot) and `loading`
is a plain boolean reporting the true state. There is nothing to narrow.

**Errors.** `loading` means **no data has arrived yet** — it is not a claim that
the subscription is healthy. A **cold-start** failure (nothing ever arrived)
leaves `loading` true *and* sets `error`, so a component that branches on
`loading` alone renders a skeleton forever; check `error` to break out of it. A
failure AFTER data arrived deliberately does NOT replace good data with an error
banner (a transient websocket hiccup would blank a working screen); those reach
the api's error bus instead — subscribe with `useOnRpcError` for
connection-level UX.



---

<!-- source: en/data/mutations.md -->
## Mutations

_`*.mutation.ts` + `*.mutation.server.ts` pairs — atomic writes with typed schemas and auto-optimistic metadata._

A **mutation** is the write primitive. Every `ctx.store.insert`, `update`, and `delete` inside one mutation runs in a single transaction. If the executor throws, the transaction rolls back and subscribers never observe a partial write.

Declare `target` metadata in the descriptor so the client can derive optimistic patches for queries whose `source` points at the same table.

Live — submitting adds the row optimistically (it shows instantly, then the
server delta confirms it):

```tsx
<AutoForm api="app" mutation="todos.create" submitLabel="Add todo" />
<DataTable api="app" query="todos.list" />
```

## Mutation Pair

Descriptor:

```ts
// apps/api/mutations/notes.create.mutation.ts
import { defineMutation } from '@voltro/protocol'
import { Schema } from 'effect'

export const createNote = defineMutation({
  name:   'notes.create',
  target: { table: 'notes', op: 'insert' },
  guards: [{ scope: 'notes:write' }],
  input:  Schema.Struct({
    title: Schema.NonEmptyString,
    body:  Schema.String,
  }),
  output: Schema.Struct({
    id: Schema.String,
  }),
})
```

> **`guards:` is what makes this file boot.** A wire-exposed mutation must
> declare exactly one of `guards:`, `openAccess: '<reason>'` or `internal: true`
> — a descriptor with none of them is refused at boot, naming the file. A write
> is also where a rubber-stamp guard costs the most, so name the scope the write
> actually needs rather than one every caller already holds. Full rules:
> [Authorization](/docs/authentication/authorization#every-procedure-decides-guards-or-openaccess).

Server executor:

```ts
// apps/api/mutations/notes.create.mutation.server.ts
import type { AppContext } from '@voltro/runtime'

export default async (
  input: { title: string; body: string },
  ctx: AppContext,
) => {
  const inserted = await ctx.store.insert('notes', {
    title:    input.title,
    body:     input.body,
    authorId: ctx.request.subject.id,
    tenantId: ctx.request.subject.tenantId,
  })

  return { id: String(inserted['id']) }
}
```

The descriptor is the wire contract. The `.mutation.server.ts` file is the server-only implementation.

`insert` returns the post-image as an untyped `Row` (`Readonly<Record<string, unknown>>`), so narrow the field you need (`String(...)`) instead of asserting it with `as string` — an assertion silences the compiler without checking anything. To read a row back, use the fluent terminals: `ctx.store.select('notes').where('id', id).one()` fails with the typed `NoRowFound` when the row is missing (or when more than one matches), so you never need a hand-written not-found branch; `.first()` / `.maybeOne()` return `null` instead. That fluent builder is string-keyed and yields an untyped `Row`; for a TYPED single row, pass the `database.<table>` builder to the store's own terminals — `await ctx.store.one(database.notes.where(eq('id', id)))` returns the row type with no cast. See [single-row terminals](/docs/database/query-builder#single-row-terminals-on-the-typed-builder).

## What The Runtime Does

1. Decode `input` with the descriptor schema.
2. Run mutation plugin interceptors.
3. Execute the server file inside `store.transactional(...)`.
4. Encode `output` with the descriptor schema.
5. Commit the transaction.
6. Drain the batched change events so matching query subscriptions receive new snapshots or deltas.

## Missing required columns fail LOUD, at the call

`ctx.store.insert` / `upsert` / `insertIgnore` check the payload against the
table before the statement runs. A column that is NOT NULL, has no default, and
isn't auto-stamped — a business FK like `teamId`, a plain `timestamp()` like
`lastRefreshedAt` — must be present, or the write raises a typed
`TableValidationFailed` naming it:

```txt
TableValidationFailed: missing required column 'lastRefreshedAt'
  — NOT NULL with no default and not auto-stamped
```

Without this the omission slips past `tsc` and boot and surfaces only as a raw
dialect `SqlError: Failed to execute statement` — and only on the INSERT path, so
it stays dormant until the first row with no existing cache entry. An
`upsert` / `insertIgnore` whose payload omits one of its own `conflictColumns` is
named the same way (an absent conflict key can't match its target). The check
runs after stamping, so auto-id, `tenantId`, and audit columns never trip it, and
it skips nullable and defaulted columns — the ones you may legitimately omit.

### Catch it at COMPILE time — `insertRow` / `upsertRow`

The runtime guard above is the backstop. To catch a missing column at compile
time, use `insertRow` / `upsertRow` — they take the **table object** (not a
string name), so the payload is typed against the table's required columns:

```ts
import { insertRow } from '@voltro/database'

// ✗ compile error — lastRefreshedAt is NOT NULL with no default
await insertRow(ctx.store, roadmapEpicStats, { teamId })
// ✓
await insertRow(ctx.store, roadmapEpicStats, { teamId, lastRefreshedAt: new Date() })
```

`InferInsertRow<T>` makes a column optional exactly when you may omit it — it's
nullable, has a `.default()`, or is a framework-filled `id` / `tenantId` / audit
column — and required otherwise. `upsertRow`'s `conflictColumns` are constrained
to the table's own columns, so a typo is a compile error too. The string-keyed
`ctx.store.insert` / `upsert` still work unchanged; the typed seam is opt-in.

> **If you adopt it, check your tooling for the old spelling.** `insertRow`
> replaces `store.insert('<tableName>', …)`, so anything that matches on that
> string stops matching — a lint rule, a codemod, an architecture test. One team
> had a test asserting every mutation writes an audit log; it matched
> `store.insert('auditLogs'`, and migrating to `insertRow` silently blinded it.
> Green suite, gap reintroduced. Grep for the old spelling before you migrate,
> not after.

## Partial updates: `ctx.store.applyDefined`

A partial-update mutation should set only the fields the caller actually sent — not overwrite an omitted field with `undefined`. Instead of hand-writing `if (input.x !== undefined) patch.x = input.x` per field, use `ctx.store.applyDefined(input, keys)`:

```ts
const execute = async (input: UpdateNote, ctx: AppContext) =>
  ctx.store.update('notes', input.id, ctx.store.applyDefined(input, ['title', 'body', 'dueAt']))
```

It returns a patch containing only the listed keys whose value is not `undefined` (a defined falsy value like `0` / `''` / `false` IS kept). Also importable standalone (`import { applyDefined } from '@voltro/runtime'`) for seeds/tests.

## Calling From React

```tsx
import { useMutation } from '@voltro/client'

export default function NewNote() {
  const create = useMutation<{ title: string; body: string }, { id: string }>(
    'app',
    'notes.create',
  )

  return (
    <form onSubmit={async (event) => {
      event.preventDefault()
      const form = new FormData(event.currentTarget)
      await create.mutate({
        title: String(form.get('title')),
        body:  String(form.get('body')),
      })
    }}>
      <input name="title" />
      <textarea name="body" />
      <button disabled={create.pending}>
        {create.pending ? 'Saving...' : 'Save'}
      </button>
    </form>
  )
}
```

`useMutation` returns `mutate`, `pending`, `error`, `data`, plus the chainable optimistic helpers.

### Handling the result — `onSuccess` / `onError` / `notify`

Pass a result handler to `mutate` instead of wrapping every call in
`try/catch/finally` + toasts. `pending` already replaces the `finally`:

```ts
const create = useMutation('app', 'teams.create')

await create.mutate(input, {
  onSuccess: (team) => setOpen(false),
  notify: { success: t('teams.created'), error: (e) => messageFor(e) },
})
```

**The load-bearing rule:** supplying an error handler (`onError` **or**
`notify.error`) marks the failure **handled** — `mutate` then resolves with
`undefined` instead of rejecting, which is what removes the `try/catch`. With no
error handler it rejects exactly as before, so an unhandled failure stays loud.
You opt in per call.

`notify` routes to an app-wide sink you register once — the framework is not
bound to any toast library:

```ts
import { setMutationNotifier } from '@voltro/client'
setMutationNotifier({ success: (m) => toast.success(m), error: (m) => toast.error(m) })
```

**The callback form is for SINGLE-SHOT writes.** A loop or a multi-step sequence
relies on the promise *throwing* to stop. Once the failure is handled the promise
resolves, so the loop cheerfully continues past the row that failed:

```ts
// WRONG — onError handles the failure, so the loop never stops
for (const row of rows) {
  await create.mutate(row, { onError: (e) => toast.error(messageFor(e)) })
}

// RIGHT — bare mutate rejects, so the sequence aborts where it broke
try {
  for (const row of rows) await create.mutate(row)
} catch (e) {
  toast.error(messageFor(e))
}
```

The same applies to `run` on [actions](/docs/data/actions).

## Idempotency — a retried mutation runs exactly once

The reactive client resends an in-flight mutation after a network blip. Without a guard, "create order" or "charge card" would run twice. `useMutation` (and `useAction`) mint a fresh **idempotency key** per call and attach it to the rpc frame; when idempotency is enabled the server dedupes a repeat of that key — the handler runs once and the retry replays the first result.

Enable it once (this also covers the REST `Idempotency-Key` header — one switch, both surfaces):

```ts
// app.config.ts
export default {
  idempotency: true, // or { ttlMs: 600_000 } — the dedup window (default 24h)
}
```

For a HIGHER-level guarantee — dedupe a double-click or an offline resend of the *same logical action* — pass a STABLE key derived from the action's identity, instead of the per-call one:

```ts
await placeOrder.mutate(cart, { idempotencyKey: `order:${cart.id}` })
```

The key is scoped to `(tenant, subject, mutation)`, so one user's key can never replay another's. The replayed result is byte-for-byte the first one — a `Date` in the output comes back a `Date`, not a string — because it round-trips through the mutation's output schema. Off by default: with no `idempotency` config, every call runs.

## Auto-Optimistic

The default path is declarative:

```ts
defineQuery({
  name: 'notes.list',
  source: 'notes',
  guards: [{ scope: 'notes:read' }],
  input: Schema.Struct({}),
  output: Schema.Array(Note),
})

defineMutation({
  name: 'notes.create',
  target: { table: 'notes', op: 'insert' },
  guards: [{ scope: 'notes:write' }],
  input,
  output,
})
```

Then the client can stay plain:

```tsx
const notes = useSubscription('app', 'notes.list', {})
const create = useMutation('app', 'notes.create')

await create.mutate({ title, body })
```

The client stages an optimistic patch, calls the server, then retires the patch. Server-pushed deltas are still the source of truth.

**Retiring the patch has exactly two shapes, and they are not interchangeable:**

- **Rollback — only when the write FAILED.** The server did not write, so the
  preview must go, immediately. Nothing else ever rolls a patch back: not a
  timer, not an elapsed window, not a heuristic.
- **Hand-off — when the server's own state arrives and reflects the write.** A
  delta always supersedes (it *is* the echo of committed writes); a fresh
  snapshot supersedes only when it actually moved the base. Base gains the real
  row and the placeholder goes in the same update, so there is no flash and no
  optimistic+real duplicate.

If a confirmed write never gets its echo — reactivity is broken for that query's
source table — the client **re-issues the subscription and keeps the preview**,
and reports it on the error bus (visible in `voltro logs`). It does not fall back
to a base it knows does not reflect the write: a user watching their saved change
disappear will either redo it or plan on a state they believe was not stored.

For special shapes, override the patch:

```tsx
const create = useMutation('app', 'notes.create').withOptimistic((cache, input) => {
  cache.forTag<ReadonlyArray<{ id: string; title: string }>>('notes.list', (rows) => [
    { id: `temp:${Date.now()}`, title: input.title },
    ...rows,
  ])
})
```

Use `.withoutOptimistic()` for effects that should not preview locally.

### Nested / path-targeted optimistic

By default a `target` patches the **flat top-level row array** a query returns, keyed by `id`. When a query returns a **nested array** — a JSON array column (`snapshot.projects`) or a computed/shaped value — add `path` (and, if the item key isn't `id`, `by`) to patch at **item** granularity, with no hand-written `.withOptimistic` reducer:

```ts
target: {
  table: 'projectRoadmaps', op: 'update',
  path: 'snapshot.projects',                 // dot-path to the nested array in the value
  identify: (input) => input.projectId,      // which item to patch (default input.id)
}
```

- `op: 'insert'` appends (or `order: 'prepend'`) a new item into the nested array — safe even on a computed query (a path insert targets a KNOWN document, not a blind top-level add).
- `op: 'delete'` filters the item out by its key.
- `by` overrides the item-key field (default `'id'`).

**Shape the item with `shapeItem` (not `shape`).** For a nested target, build/patch the item with `shapeItem` — it is typed to the **item** of the nested array, not the mutation's output, so `current` needs no cast:

```ts
target: {
  table: 'projectRoadmaps', op: 'update', path: 'snapshot.projects',
  identify: (input) => input.projectId,
  shapeItem: (input, current) => ({ ...current, startDate: input.startDate }),  // `current` IS the item
}
```

(The flat `shape` stays bound to the output row — a single field can't be both, so the nested shaper is its own.)

**Bulk (multi-item) patches.** `identify` may return an **array** of ids to patch or delete **many** items in one mutation — exactly the group-drag / batch-edit where per-item parallel writes used to race:

```ts
target: {
  table: 'projectRoadmaps', op: 'update', path: 'snapshot.projects',
  identify: (input) => input.projectIds,                    // ← ARRAY: patch them all
  shapeItem: (input, current) => ({ ...current, shiftedBy: input.delta }),  // each keeps its own key
}
```

(This works for flat top-level targets too — `identify` returning an array patches/deletes every matching row.)

Add `match` to patch **only** the entries whose current value satisfies a predicate — the guard that stops a patch bleeding across sibling subscriptions sharing a source table:

```ts
target: {
  table: 'projectRoadmaps', op: 'update', path: 'snapshot.projects',
  identify: (i) => i.projectId,
  match: (value, input) => value.id === input.roadmapId,   // only THIS roadmap's subscription
}
```

`path`, `by`, `match`, and `shapeItem` are browser-safe descriptor data (a dot-path string + pure functions) — the same discipline as `identify`/`shape`.

## Typed Errors

```ts
import { Schema } from 'effect'

class NoteQuotaExceeded extends Schema.TaggedError<NoteQuotaExceeded>()('NoteQuotaExceeded', {
  limit: Schema.Number,
}) {}

export const createNote = defineMutation({
  name: 'notes.create',
  target: { table: 'notes', op: 'insert' },
  guards: [{ scope: 'notes:write' }],
  input,
  output,
  error: NoteQuotaExceeded,
})
```

Throw a matching error from the server file; the client can narrow on `_tag`.

### Matching typed errors on the client

Tagged errors **round-trip structurally** over the wire — the caught value carries `_tag` plus every declared field as real properties (and `instanceof` works, same Schema class both ends). You do **not** need to parse the error message string.

Inside Effect, use `Effect.catchTag('NoteQuotaExceeded', …)`. In a React `try/catch` (outside Effect, where `catchTag` isn't available and the decoded value may be a plain object, not a class instance), match with **`errorTag(err)`** — the dependency-free tag reader `@voltro/protocol` ships (it reads what `toRpc` writes):

```ts
import { errorTag } from '@voltro/protocol'

try {
  await createNote(input)
} catch (err) {
  if (errorTag(err) === 'NoteQuotaExceeded') {
    // err.limit is the declared field — read it directly, no regex
  }
}
```

**Trace id for debugging.** An error caught from `useMutation` / `useAction` carries a **non-enumerable `__voltroTraceId`** — the bridge to the server logs for that exact call:

```ts
const traceId = (err as { __voltroTraceId?: string }).__voltroTraceId
// → `voltro logs --trace <traceId>` to see the server-side span
```

`errorTag` lives in `@voltro/protocol` rather than in the client for a reason
worth knowing before you decide where your own error handling goes: **`_tag` is a
wire concept, and protocol owns the wire.** So a shared error handler in a package
that has no business depending on `@voltro/client` — a UI kit, an i18n layer —
can read a tag without taking that dependency. `instanceof` is the thing that
does NOT survive the wire: what arrives in the browser was decoded from JSON and
never constructed, so match on the tag, not on the class.

**Exhaustive matching with the generated `matchError`.** Codegen emits a per-app `matchError` (plus `AppError` / `AppErrorTag`) into `rpcGroup.generated.ts`, derived by reference from every descriptor's `error:` schema + your plugins' cross-cutting errors — so there's no hand-maintained tag list to drift out of date (a dead/renamed tag is a compile error):

```ts
import { matchError } from './rpcGroup.generated'

const message = matchError(err, {
  NoteQuotaExceeded: (e) => `Limit ${e.limit} reached`,   // e is typed
  ScopeError: (e) => `Missing ${e.required}`,
}, () => 'Something went wrong')
```

## Cross-table business rules — `.rule()`

A [typed error](#typed-errors) is declared on ONE mutation. A **rule** is declared on a TABLE: an invariant the runtime enforces on EVERY mutation that writes that table, no matter which one did the write. Declare it in the schema with `.rule(name, predicate)`:

```ts
import { table, id, integer, eq } from '@voltro/database'
import { database } from '../database/schema'

export const invoices = table('invoices', {
  id:    id(),
  total: integer(),
}).rule(
  'totalMatchesLineItems',
  async (row, { store }) => {
    const items = await store.query(
      database.lineItems.where(eq('invoiceId', row.id)).descriptor,
    )
    const sum = items.reduce((acc, l) => acc + Number(l.amount), 0)
    return sum === row.total || { params: { computed: sum, declared: row.total } }
  },
)
```

The predicate receives the **post-write row** and a `context` whose `store` is the SAME transactional store the mutation wrote through — so a cross-table read shares the write's MVCC snapshot and cannot race it. It runs INSIDE the mutation transaction, after the write and before commit, and it is dialect-neutral (the same predicate is correct on all four dialects — no per-dialect code). Return `true` (or nothing) when the invariant holds; return `false` or a `RuleViolationDetail` (`{ params?, field?, message? }`) to signal a violation for THIS row.

Unlike [`.check()`](/docs/database/columns#db-level-checks) — a single-row SQL `CHECK` the DATABASE enforces as DDL — a rule is a PREDICATE the runtime evaluates, so it can read OTHER tables. Use `.check()` when the database itself must guarantee a single-row constraint; reach for `.rule()` for a cross-table invariant (an invoice total matching its line items, a booking not exceeding a resource's capacity).

### The violation is a typed error — automatically

A violated `error`-severity rule rolls the whole mutation back and fails with the typed, wire-preserved **`BusinessRuleViolation`**. You do **not** declare it on the mutation's `error:` — the runtime auto-merges it into every mutation's error union at the wire boundary, exactly like `ScopeError`. A rule declared on a SCHEMA table can fail ANY mutation that writes that table, so no single descriptor could know to declare it; the auto-merge is what keeps the violation a typed error the client decodes by `_tag` rather than an untyped defect.

Match it on the client the same way as any [typed error](#matching-typed-errors-on-the-client):

```ts
import { errorTag } from '@voltro/protocol'

try {
  await createInvoice(input)
} catch (err) {
  if (errorTag(err) === 'BusinessRuleViolation') {
    // err.rule — the rule name; err.params — the offending values; err.field — the pointer
  }
}
```

`BusinessRuleViolation` carries `rule` (the declared name), optional `params` (i18n params for the offending values), an optional `field` pointer, and `severity` (always `'error'` on the wire — a warning-severity rule never reaches the client as an error).

### `severity: 'warning'` — log without blocking

```ts
export const invoices = table('invoices', { id: id(), total: integer() }).rule(
  'totalMatchesLineItems',
  async (row, { store }) => true,
  { severity: 'warning' },
)
```

A `warning`-severity rule LOGS and audits the violation but lets the write commit — useful while backfilling data that does not yet satisfy a newly-added invariant. The default is `'error'` (roll back).

Rules run on `voltro dev` and `voltro serve` through the same mutation runner, so the two boot paths cannot disagree about whether a rule fires. Inserts and updates on your own tables are re-validated; a delete is not (there is no post-write row to check).

## `internal: true` — off the wire entirely

Every discovered `*.mutation.ts` / `*.query.ts` / `*.action.ts` / `*.stream.ts`
gets a route on the WebSocket — which is why each one has to declare **who may
call it**, and why an app carrying an undecided procedure does not boot at all.
`internal: true` is the third answer to that question, beside `guards:` and
`openAccess:`: there is no wire surface to make a decision about, because the
procedure never gets a route. (`publicApi` and `exposeAsTool` go the other way
and opt IN to *wider* surfaces — which is why neither combines with `internal`.)

Reach for it when the caller is other **server** code — a workflow step, a
schedule, another executor — and never a browser:

```ts
export const createFromAction = defineMutation({
  name: 'auditLog.createFromAction',
  input: Schema.Struct({ actorId: Schema.String, eventType: Schema.String }),
  output: Schema.Void,
  internal: true,
})
```

It is not emitted into `rpcGroup.generated.ts`, and neither `voltro dev` nor
`voltro serve` registers a route — the tag is unroutable over `/rpc` and the
WebSocket. Server code calls it by importing its executor directly.

**A naming convention is not a boundary.** One app had grown 18 procedures named
`*Internal`, meaning "only other server code calls this"; all 18 were in the
client group, and one of them accepted `actorId` / `actorEmail` / `actorType`
from the caller and wrote an audit row. No guard, zero callers, reachable by
anyone logged in. If the only thing keeping a procedure off the wire is that
nobody wrote a client call for it, it is on the wire — the same reasoning as
`.serverOnly()` on a column, one level up.

That app is why the boot gate exists: all 18 declared no access decision, so
today it does not start until each of them says `guards:`, `openAccess:` or
`internal: true`. The gate turns "reachable and nobody looked" into a refusal
naming every file — but it only forces the question, it cannot answer it, and
`internal: true` is the right answer only when no browser is meant to call the
procedure at all.

**It is not a substitute for a guard.** An internal procedure still runs with
whatever authority its caller has. This removes the wire surface, not the need to
check who is asking; `voltro doctor`'s authz scan still covers it.

**It cannot be combined with `publicApi` or `exposeAsTool`.** Those add a REST
route and an agent tool respectively — opt-ins to a *different* surface — so a
procedure carrying both would be unreachable from your own client and reachable
from the internet. That combination throws where it is declared:

```text
auditLog.createFromAction: `internal: true` cannot be combined with `publicApi`.
```

Neither silent resolution would be right: dropping the REST route breaks a live
endpoint invisibly, and keeping it defeats the flag. Drop `internal: true` if the
wider surface is intended, or remove the annotation if it is not.

The flag also removes the procedure from `voltro dev`'s inspect invoker, so the
devtools "invoke" panel will not list it. That is deliberate — an internal
procedure is the one most likely to carry no guard, since "only server code calls
this" is the reason people write them.

## When Not To Use A Mutation

- **External I/O.** Use an action or workflow.
- **Progress output.** Use a stream.
- **Reads.** Use a query.
- **Long-running durable work.** Use a workflow.



---

<!-- source: en/data/actions.md -->
## Actions

_`*.action.ts` + `*.action.server.ts` pairs — unary server calls for external I/O and non-transactional work._

An **action** is a typed unary RPC that runs outside the mutation transaction wrapper. Use it for external I/O and request-scoped work: HTTP calls, emails, signed upload URLs, AI calls that return one value, or kicking off a workflow.

Actions can read or write through `ctx.store`, but those writes are not grouped into one automatic transaction and they do not drive client auto-optimistic updates. If the main purpose is an atomic database write, use a [mutation](/docs/data/mutations). If the client should receive incremental elements while work is running, use a [stream](/docs/data/streams).

Live — a unary action: call it, get one typed answer back. No transaction, no
optimistic patch:

```tsx
const echo = useAction('app', 'demo.echo')
await echo.run({ message })   // one call → one typed result
```

### Actions declare what they touch

An action is non-transactional external I/O, and it often touches a table on the
way — a cache it fills, a job row it stamps. Declare it the same way a query and
a mutation do:

```ts
export const syncIssue = defineAction({
  name: 'jira.syncIssue',
  guards: [{ scope: 'jira:sync' }],
  input: Schema.Struct({ key: Schema.String }),
  output: Schema.Struct({ ok: Schema.Boolean }),
  source: 'jiraIssueCache',
  target: { table: 'jiraIssueCache', op: 'upsert' },
})
```

Without it the table is invisible to `voltro check`, which then reports it as an
orphan. While any action declares neither, the orphan rule says so and asks for
the declaration rather than proposing you delete the table — a wrong finding
whose remedy is destructive is worse than a wrong finding.

## Action Pair

Descriptor:

```ts
// apps/api/actions/support.ping.action.ts
import { defineAction } from '@voltro/protocol'
import { Schema } from 'effect'

export const pingExternal = defineAction({
  name: 'support.ping',
  guards: [{ scope: 'support:diagnostics' }],
  input: Schema.Struct({ url: Schema.String }),
  output: Schema.Struct({
    status:     Schema.Number,
    durationMs: Schema.Number,
  }),
})
```

> **`guards:` is what makes this file boot** — a wire-exposed action must declare
> exactly one of `guards:`, `openAccess: '<reason>'` or `internal: true`, or the
> boot refuses it, naming the file. This one earns a real scope rather than
> `openAccess`: it fetches a **caller-supplied URL** from your server. The
> [SSRF guard](/docs/security/overview#outbound-http-is-ssrf-guarded-by-default)
> keeps that off your internal network, but "anyone on the internet may make this
> server issue requests" is still not a claim to make by accident. Full rules:
> [Authorization](/docs/authentication/authorization#every-procedure-decides-guards-or-openaccess).

Server executor:

```ts
// apps/api/actions/support.ping.action.server.ts
import { HttpClient, HttpClientRequest } from '@effect/platform'
import { Effect } from 'effect'

export default (input: { url: string }) =>
  Effect.gen(function* () {
    const client = yield* HttpClient.HttpClient
    const startedAt = Date.now()
    const response = yield* client.execute(HttpClientRequest.get(input.url))

    return {
      status: response.status,
      durationMs: Date.now() - startedAt,
    }
  })
```

The descriptor file is safe for client-side discovery. The `.action.server.ts` file contains the server-only code and can return a plain value, a `Promise`, or an `Effect`.

## Calling From React

```tsx
import { useAction } from '@voltro/client'

const ping = useAction<{ url: string }, { status: number; durationMs: number }>(
  'app',
  'support.ping',
)

const onClick = async () => {
  const result = await ping.run({ url: 'https://example.com/health' })
  console.log(result.status)
}
```

`useAction` returns `run`, `pending`, `error`, and `lastResult`.

### Handling the result — `onSuccess` / `onError` / `notify`

`run` takes the same options bag as a [mutation](/docs/data/mutations)'s `mutate`,
so a one-shot action does not need a `try/catch/finally` + toast wrapper —
`pending` already replaces the `finally`:

```ts
const invite = useAction('app', 'invites.send')

await invite.run({ email }, {
  onSuccess: (out) => toast.success(`sent ${out.id}`),
  onError:   (e)   => toast.error(readError(e)),
})
```

**The load-bearing rule:** supplying an error handler (`onError` **or**
`notify.error`) marks the failure **handled** — `run` then resolves with
`undefined` instead of rejecting, which is what removes the `try/catch`. With no
options it rejects exactly as before, so an unhandled failure stays loud.

**The callback form is for SINGLE-SHOT calls.** A loop or a multi-step sequence
relies on the promise *throwing* to stop; once the failure is handled the promise
resolves and the sequence runs on past the step that broke:

```ts
// WRONG — onError handles the failure, so the loop never stops
for (const email of emails) {
  await invite.run({ email }, { onError: (e) => toast.error(readError(e)) })
}

// RIGHT — bare run rejects, so the sequence aborts where it broke
try {
  for (const email of emails) await invite.run({ email })
} catch (e) {
  toast.error(readError(e))
}
```

### Multi-step sequences — `useSequence`

The rule above says a loop keeps its `try/catch`, and that left multi-step
writes as the only verbose thing on the write path — while being the hardest
case, not the easiest. `useSequence` runs the steps in order, gives the whole
sequence **one** `onError`, and lets each step say how to undo itself:

```tsx
const seq = useSequence({ onError: (e) => toast.error(readError(e)) })

const result = await seq.run(
  sequence()
    .step('upload', () => upload.run({ file }), {
      undo: (created) => removeObject.run({ id: created.id }),
    })
    .step('attach', (c) => createAttachment.run({ refId: c.upload.id })),
)

if (result.ok) setAttachmentId(result.data.attach.id)
```

Each step's context is typed and accumulates, so a fourth step can read the
first step's result by name. `run` resolves with a discriminated result rather
than rejecting — same "handled" semantics as `useMutation`/`useAction`.

**`undo` receives its own step's result**, which is the point: the id you just
created is what you need to delete it again. Undos run in reverse for the steps
that already succeeded.

**This is not a transaction, and the difference matters.** After `createTicket`
returns, the ticket exists in Jira; nothing the browser does un-creates it, it
can only issue a delete and hope. And the compensation runs *in the tab* — close
it, lose the network, or navigate away mid-rollback and the remaining undos
never happen. Two rules follow, and the primitive enforces both:

- the step that **failed** is never compensated (it may or may not have had an
  effect — undoing it is a guess, and a wrong guess deletes something else);
- a failing `undo` never replaces the original error, and never stops the
  remaining undos. Cleanup failures come back in `compensationFailures`, so an
  orphan is something you can see rather than something you find later.

**Undos are independent, so keep them idempotent — or say they overlap.** Every
succeeded step's undo runs, and the runner cannot tell whether two of them
reverse the same thing. If `deleteJiraDraftTicket` deletes the issue *and*
discards the draft, the earlier `discardDraft` undo runs on something already
gone. Harmless when discarding is idempotent; a real defect for a refund or a
cancellation email. Declare the overlap instead of relying on luck:

```tsx
sequence()
  .step('draft', () => createDraft.run(input), { undo: (d) => discardDraft.run({ id: d.id }) })
  .step('jira',  (c) => createTicket.run({ draftId: c.draft.id }), {
    undo:   (t, ctx) => deleteJiraDraftTicket.run({ key: t.key, draftId: ctx.draft.id }),
    covers: ['draft'],
  })
```

An `undo` receives `(result, ctx)` — its own result *and* the context up to that
step, the same context `covers` and `when` see. An inverse that needs an id from
an earlier step (here `deleteJiraDraftTicket` wants both the jira key and the
`draftId`) reads it from `ctx` rather than the step re-returning it just so the
undo can reach it. A later step's result is not in `ctx`: it is rolled back
before this one, so reading it would be reading something already reversed.

If the covering undo *fails*, the covered steps are neither run nor claimed —
whether the cascade got that far is unknown, and both guesses are wrong. They
come back in `compensationUncertain` so you can reconcile.

**One optional step is fine; a loop or a real branch is not.** `when` skips a
step, and a skipped step contributes `undefined` to the context (the type says
so) and gets no undo:

```tsx
sequence()
  .step('save', () => save.run(input))
  .step('summary', (c) => scheduleSummary.run({ id: c.save.id }), {
    when: (c) => c.save.changed,
  })
```

That is deliberately narrow: it catches "linear except for one `if`", which is a
different shape from a loop. Loops and full branches still keep their
`try/catch`.

**When the rollback has to survive a closed tab, this is the wrong tool.** Put
the sequence in a [workflow](/docs/workflows/overview): the engine owns retries
and compensation there, and a crash resumes instead of leaking.

## Action vs Mutation vs Stream

| Need | Use |
|---|---|
| Atomic database write with rollback | Mutation |
| External I/O or one-shot server call | Action |
| Progressive server-to-client output | Stream |
| Durable multi-step work | Workflow |
| Client-side optimistic preview | Mutation with `target` metadata |

## Database Writes

Actions are not wrapped in the mutation transaction. A throw does not roll back previous writes or external side effects.

```ts
export default async (input, ctx) => {
  await ctx.store.insert('auditLogs', {
    message: input.message,
    actorId: ctx.request.subject.id,
  })

  await sendExternalWebhook(input)
  return { ok: true }
}
```

Use this shape only when that ordering is acceptable. If several writes must commit or roll back together, move them to a mutation. If the external side effect must survive retries, model the operation as a workflow.

## Typed Errors

Actions use the same `error: Schema.Union(...)` descriptor field as mutations. Throwing a matching tagged error reaches the client as a typed failure from `run(...)`.

See [Error handling](/docs/data/errors) for the pattern.

## Anti-Patterns

- **Actions used as reads.** Use a query if the client wants cached reactive data.
- **Actions used for atomic writes.** Use a mutation so subscribers never see partial state.
- **Actions used for progress feeds.** Use `defineStream` and `useAgentStream`.



---

<!-- source: en/data/subscriptions.md -->
## Subscriptions

_How reactive query subscriptions stay live over WebSocket._

A **subscription** is what the browser gets when it calls `useSubscription(...)` for a `defineQuery` RPC. The app code writes a query pair; the runtime keeps that query live over WebSocket and pushes new snapshots or deltas when matching data changes.

Every table is reactive by default, so a query over any table is live with
nothing to configure. A table explicitly marked
[`.nonReactive()`](/docs/database/overview) emits no change events at all — a
subscription over one returns its first snapshot and then stays silent forever,
which is why `voltro dev` warns about that combination at boot.

Subscriptions are not a separate file convention anymore. The file convention is [queries](/docs/data/queries): `*.query.ts` for the descriptor and `*.query.server.ts` for the executor.

Live — add a todo (or open this page in a second tab) and the list updates with
no refetch; the `<DataTable>` is a subscription under the hood:

```tsx
<AutoForm api="app" mutation="todos.create" submitLabel="Add todo" />
<DataTable api="app" query="todos.list" />   {/* live subscription */}
```

## Lifecycle

```tsx
const { data } = useSubscription('app', 'notes.list', { archived: false })
```

1. The client opens or reuses the API WebSocket.
2. It sends the query tag and input.
3. The runtime runs `notes.list.query.server.ts`.
4. The first value is delivered as a snapshot.
5. Later mutations emit change events when their transaction commits.
6. Matching query subscribers receive the updated value.

From React, `data` just changes. There is no `refetch` call.

## Query Descriptor

```ts
// apps/api/queries/notes.list.query.ts
import { defineQuery } from '@voltro/protocol'
import { Schema } from 'effect'

export const listNotes = defineQuery({
  name:   'notes.list',
  source: 'notes',
  guards: [{ scope: 'notes:read' }],   // re-checked on every delivery, not just at open
  input:  Schema.Struct({ archived: Schema.Boolean }),
  output: Schema.Array(Schema.Struct({
    id:    Schema.String,
    title: Schema.String,
  })),
})
```

`source` declares which table re-runs computed queries and lets mutations with matching `target` metadata patch the client cache optimistically.

## Reactivity channels

`source:` usually names a table. It does not have to, and when the state a query
reads is **not in the database**, naming one is the wrong answer.

```ts
import { defineQuery, reactivityChannel } from '@voltro/protocol'
import { Schema } from 'effect'

// Declared once, in a browser-safe module both sides import.
export const jobQueue = reactivityChannel('job-queue')

export const queueDepth = defineQuery({
  name:   'jobs.depth',
  source: jobQueue,                     // ← a channel, not a table
  input:  Schema.Struct({}),
  output: Schema.Struct({ pending: Schema.Number }),
})
```

Then push it from wherever the state actually changes:

```ts
import { publishReactivity } from '@voltro/protocol'

publishReactivity(ctx.store, jobQueue)   // every subscriber re-runs its executor
```

Everything else is unchanged: the executor returns a plain value, the framework
re-runs it and pushes the result over the same subscription transport a
table-backed reactive query uses. There is no second client concept and no
second push mechanism — `useSubscription` does not know the difference.

### Why not just declare a table

Because the alternatives are worse, and the framework shipped one of them for a
release. Presence held its roster in memory and still declared a
`_voltro_presence` table it never wrote a row to, purely to own a name the
reactivity layer would route on — an empty table in every user's database,
created by every migration and diffed on every boot.

The other tempting option is to point `source:` at a name that resolves to
nothing. That is worse than the empty table: the [stale-`source` boot
warning](#fan-out--how-many-subscribers-may-one-change-wake) is the only signal
for a subscription that has gone permanently quiet, and an exemption for a name
you invented disables it for the one case it was built for.

### A table `source:` is typechecked

`voltro dev` writes `voltro-tables.generated.d.ts` beside your generated rpc group, listing every table the app has — your entities, your plugins' `extendSchema.tables`, and the framework's own. `source:` is narrowed to those names, so a typo or a table you renamed away is a **compile error**:

```ts
source: 'task_subtasks',   // ✗ Type '"task_subtasks"' is not assignable to type 'TableName'
source: 'task_sub_tasks',  // ✓
```

That matters because the failure it replaces is silent. A `source:` is matched by NAME against change events, so one that matches nothing does not break the query — it makes it never update. The write still lands, a reload still shows it, and the panel keeps showing the old value.

The file is generated, so **commit it** like the rpc group and let `voltro dev` rewrite it. Before the first run — and in a project that never generates it — `source:` is plain `string` again, which is exactly the previous behaviour; there is no configuration and nothing to opt into.

The error carries a suggestion when the name is close to a real one — `Did you mean '"error_logs"'?` — so a rename usually resolves without leaving the editor.

#### A COMPUTED `source:` can keep its names

A generic reader — the table arrives in `input`, the caller picks it out of a registry — is computed, and yet every name it can produce is known. The obvious derivation is not assignable:

```ts
source: Object.keys(JUNCTION_REGISTRY),   // string[] — ✗
```

The tempting exit is a cast, or annotating `ReactivitySourceValue` (the wide `string | ReactivityChannel` shape, exported from `@voltro/protocol`). Both compile at once and take that **entire** set of tables out of the check permanently. Keep the literal's keys instead:

```ts
const REGISTRY = { … } as const satisfies Readonly<Record<string, JunctionMeta>>

// consumers that index with a plain string still get the wide type
export const JUNCTION_REGISTRY: Readonly<Record<string, JunctionMeta>> = REGISTRY

// … and the source list keeps its literal names
export const JUNCTION_TABLE_NAMES = Object.keys(REGISTRY) as ReadonlyArray<keyof typeof REGISTRY>
```

The part that is easy to get wrong: an **annotation widens the keys back**, even when the literal carries `as const`. A registry written as `export const R: Readonly<Record<string, Meta>> = { … } as const` has `keyof typeof R === string`, and nothing about it looks wrong — the annotation is checked against the literal and then replaces its type. `satisfies` checks without replacing. That is the whole reason for the two-line split above.

Reach for `ReactivitySourceValue` when there genuinely is no key set to keep — a name read from a config file, or assembled at runtime. Not when recovering one takes two lines.

Two things it deliberately does not narrow. A **plugin's** route `source:` stays `string`: a plugin ships against many apps and cannot know any of their tables. And nothing that READS a descriptor's source at runtime narrows either — a reader that refused an unknown name would reject the stale name it exists to report.

**It does not check the other direction — except for relations.** A `source:` that omits a table the query genuinely reads is silent: the name is right, the table exists, and nothing has an opinion. That is the failure that costs a user report — they type, the row lands, and the panel does not move.

`voltro doctor` closes the part of it that can be closed without guessing:

```
✗  1 query loads a relation it does not declare:
   tasks.getById: eager-loads `subTasks` from 'tasks' but does not declare
   'task_sub_tasks' in `source:` — the view will not update when 'task_sub_tasks' changes.
```

An eager-loaded relation is composition by definition — its rows are IN the result — and its table comes from the relation registry, so the rule has no heuristic and no exception list. A **many-to-many** wants the junction table too, and says so separately: adding or removing a link writes only the junction row, so declaring the target alone leaves the list stale on exactly the operation a user performs to change it.

Nested relations resolve in the **same** pass: a relation loaded under `.with({ subTasks: { with: { watchers: true } } })` is looked up on `task_sub_tasks`, not on the query's base, so both levels are reported at once rather than one per run.

The one thing it still cannot see is a base table it cannot find — neither declared in `source:` nor written as `database.<name>` / `.select('<name>')`. There is then no table to resolve the relations against, and the query would otherwise report clean while every relation on it is unchecked. So doctor says so:

```
⚠  1 query could not be fully checked — the base table is neither declared nor readable from the executor:
   reports.byKind: `blocked` resolved against no table
   declare the base table in `source:` and re-run — relations under it are unchecked until then.
```

What doctor deliberately does NOT do is derive every table an executor reads. That needs a judgement — does this read **compose** the result or merely **restrict** it? — and only composing reads belong in `source:`; a restricting one re-running on every unrelated write puts a hundred lists back on the wire. A scan has to infer that from syntax, and a rule that guesses on a correct codebase teaches you to ignore it.

So: when a live view does not update, check the writing table is in the reading query's `source:` before anything else.

### Pass the channel, not its key

A channel's routing key is `channel:<name>`, and you can read it off
`jobQueue.key`. **Do not write that string into `source:`.** Passing the object
creates an import edge from the query to the declaration, which removes the
entire class of stale-source bugs for channels: a channel that is not imported
does not exist to be named. The boot audit reports a
`channel:` key nothing declared, for the two ways round it.

Names are lowercase kebab segments separated by dots — `presence`, `job-queue`,
`billing.usage`. A `:` is refused (it is the namespace separator) and so is an
uppercase letter (a key that differs only by case reads as one channel and
routes as two). Declaring the same name twice returns the same channel.

### A publish is LOCAL news, and says so

The change event a publish emits carries `origin: 'inline'` — the same stamp a
write made in this process carries. An event that arrives from another replica
carries `origin: 'injected'`. So a listener can tell the two apart:

```ts
store.onChange((event) => {
  if (event.table !== jobQueue.key) return
  if (event.origin === 'injected') {
    // a PEER published — mirror it into local state, then publish onward if
    // this node has anything of its own to add
  }
})
```

That distinction is what makes a fan-out pattern writable. Without it, a
replica's own publish and a peer's are indistinguishable, and the only way to
avoid an echo is a boolean per channel.

Two consequences worth knowing:

- **Re-publishing from inside a change listener works, synchronously.** A
  channel published while handling a peer's event reaches the other replicas
  like any other publish. (This is worth stating because it did not use to:
  `plugin-broadcast` suppressed every emission made while it was injecting, so
  such a publish woke the local node and silently never left it.)
- **`origin` never crosses the wire.** It describes how an event reached *this*
  process, so the receiving replica always stamps its own. A publish that says
  `'inline'` here arrives as `'injected'` there, which is the truth on both ends.

### What a channel is not

- **Not an event.** [`defineEvent`](/docs/data/subscriptions#streams-are-different)
  carries a PAYLOAD to subscribers with replay, ordering and gap detection. A
  channel carries nothing — it says "re-read", and the subscriber's own executor
  decides what that means. Reach for an event when the message matters; a channel
  when the state does.
- **Not cross-replica.** `publishReactivity` wakes subscribers on the node that
  called it. Fanning a change out to other replicas is
  [`@voltro/plugin-broadcast`](/docs/plugins/broadcast)'s job, exactly as it is
  for table changes on a dialect without CDC.
- **Not free per subscriber.** A publish wakes every subscriber of that channel
  and re-runs each one's executor; the channel is one routing key, so
  subscribers looking at different slices of the state are woken too. Publish on
  a real change, not on a timer — see [Fan-out](#fan-out--how-many-subscribers-may-one-change-wake).

## Query Executor

```ts
// apps/api/queries/notes.list.query.server.ts
import { eq } from '@voltro/database'
import { database } from '../database/schema'

export default (input: { archived: boolean }) =>
  database.notes.where(eq('archived', input.archived)).orderBy('createdAt', 'desc')
```

Descriptor-returning executors get fine-grained row matching. Computed-return executors can return arrays, objects, scalars, or `null`; they should declare `source` so the runtime knows what table should re-run them.

## `skip`

Use `{ skip }` when the input is not ready yet:

```tsx
const { data } = useSubscription(
  'app',
  'messages.list',
  { channelId },
  { skip: channelId === undefined },
)
```

While skipped, no WebSocket subscription opens and `data` stays `undefined`.

### A failed cold start is `failed`, not `loading`

If the first snapshot never arrives because the subscription ERRORED, the state
is **`failed: true`, `loading: false`, `data: undefined`** — nothing is in
flight and nothing more is coming, so "it is loading" would be a false
statement. `error` carries the cause.

```tsx
const s = useSubscription<Team[]>('app', 'teams.list')

if (s.loading) return <Skeleton/>
if (s.failed)  return <RetryPanel error={s.error}/>
return <TeamsTable teams={s.data}/>
```

`failed` is a positive check on purpose. The state used to be `loading: true`
with `error` set, and the escape hatch was reading that second field — which the
natural shape of a wrapper (`{ data, loading }` passed through) drops. **If you
wrap this state, carry `failed` with it**, or your callers inherit an infinite
skeleton through your hook.

Two things this does NOT cover. A failure AFTER the first snapshot leaves the
good data on screen and sets `error` — replacing working rows with an error
because the socket hiccuped is worse than the hiccup. And the failed state is
terminal for one TRANSPORT, not forever: a reconnect discards the error and
re-subscribes, so the entry returns to loading on its own.

### Skipped is `idle`, not `loading`

A skipped subscription reports **`idle: true`, `loading: false`**. The two are
different questions — "waiting for the first snapshot" and "not asking at all" —
and conflating them breaks the pattern this hook otherwise blesses:

```tsx
// WRONG on a skipping call site — renders a skeleton for a query you switched off
if (loading) return <Skeleton/>
```

Passing a dynamic `skip` therefore changes the return type: you get a third
state, and TypeScript will not let you ignore it.

```tsx
const s = useSubscription<Team[]>('app', 'teams.list', {}, { skip: !open })

if (s.idle)    return null          // deliberately not asking
if (s.loading) return <Skeleton/>   // asking, no answer yet
return <TeamsTable teams={s.data}/> // `data` is Team[] — narrowed
```

Call sites that never skip are untouched — `if (loading)` still proves `data` is
present there, and a literal `{ skip: false }` counts as never skipping. The
third state exists only where "not asking" is a real outcome.

With a `fallback` there is nothing to narrow either way: `data` is always
present, and `idle` tells you whether what is on screen is the fallback because
you chose not to ask.

A subscription that was live and is then skipped goes idle — it does **not**
keep serving the snapshot it still holds. Otherwise `skip: !open` would show
last time's data the moment a dialog reopens.

## SSR preload — first paint with data

By default a `useSubscription` on an SSR page flashes its empty/loading state on
mount, THEN opens the WebSocket and fetches the first snapshot — even though the
server could have fetched that value during the render. **`usePreloadedSubscription`**
closes that gap: it reads its first value from the SSR hydration payload, renders
real data on the first paint, then upgrades to the live stream the instant its
first snapshot lands.

Two things wire it up: the hook, and a `preload` export on the page.

```tsx
import { usePreloadedSubscription } from '@voltro/client'

export const preload = ['notes.list']

export default function NotesPage() {
  const { data } = usePreloadedSubscription<Note[]>('app', 'notes.list')
  // `data` is present on the first paint — no loading flash
  return <NotesTable notes={data} />
}
```

`export const preload` lists the subscriptions the page's tree needs at first
paint. During the SSR render (`voltro dev` and `voltro start`) the framework runs
each one server-side — the SAME `ctx.query(tag, input)` a loader receives — and
seeds the result into the hydration payload. The client's
`usePreloadedSubscription` finds the seed by the SAME cache key `useSubscription`
uses, so the server markup and the client hydration render read an identical
value — there is no hydration mismatch. The live subscription still opens and
takes over; the seed is only the first value, never the source of truth.

### `preload` entries

An entry is either a bare tag or a tag plus an `input` derived from the route
params:

```tsx
export const preload = [
  'teams.list',                                                    // no input
  { tag: 'project.detail', input: (params) => ({ id: params.id }) },
]
```

The `input` you derive here MUST match the `input` you pass the hook — both
address the same cache entry:

```tsx
export default function ProjectPage({ params }: { params: { id: string } }) {
  const { data } = usePreloadedSubscription('app', 'project.detail', { id: params.id })
  return <ProjectView project={data} />
}
```

### Falls back to `useSubscription`

When no seed exists for the key — a client-side SPA navigation the server never
rendered, or the static prerender (which has no live api origin) —
`usePreloadedSubscription` behaves EXACTLY like `useSubscription`: it loads until
the stream answers. So it is always safe to reach for; the preload is a
first-paint optimization, never a correctness dependency. A failed preload is
likewise non-fatal — the live subscription still delivers the value on the
client, the only loss is the first-paint seed.

### `preloadFailed` — "no data" vs "could not get data"

A preload that FAILED server-side and one that was never declared arrive the same
way: as the absence of a seed. That makes an empty first paint ambiguous, and the
ambiguity is not academic — a session cookie that has outlived the IdP's token
lifetime makes EVERY preload on the page fail at once, so the page renders its
empty state while the server log holds the only explanation.

`preloadFailed` separates the two:

```tsx
const projects = usePreloadedSubscription<Project[]>('app', 'projects.list')

if (projects.loading) return <Skeleton/>
if (projects.preloadFailed) return <Spinner label="Loading…"/>  // not empty — unasked
return <ProjectTable rows={projects.data}/>
```

It is a boolean and says nothing about WHY. The server's failure text is a
refused call's error message; it belongs in the server log, which is the one
place a browser cannot read. It also says nothing about the LIVE subscription,
which usually recovers on its own — the browser reconnects with a credential the
SSR request did not have. Read it as "the first paint has no server data, and
that was not for lack of asking", which is exactly enough to pick a spinner over
an empty state.

### Seeding by hand

`export const preload` is sugar over an explicit seed. When a loader ALREADY has
the value — you fetched it for the `<title>`, a breadcrumb, or the row name — seed
it directly with `seedPreloadedSubscription` (server-side only) instead of
fetching it a second time:

```ts
import { seedPreloadedSubscription } from '@voltro/client'

// inside a loader / layout loader, server-side
const notes = await ctx.query('notes.list', {})
seedPreloadedSubscription('app', 'notes.list', {}, notes)
```

Calling it outside a server render throws — on the client the live subscription
already provides the value, so a client-side seed would be meaningless.

## Streams Are Different

For non-database or transient element feeds, use [streams](/docs/data/streams), not subscriptions:

```tsx
const ticker = useAgentStream('app', 'ticker.watch')
ticker.start({ symbol: 'BTC' })
```

Queries/subscriptions are for live state. Streams are for one-shot element flows such as tokens, progress events, and import logs.

## Reconnect

A dropped WebSocket rebuilds the whole client stack — new socket, new RPC
client, new subscription cache — and re-subscribes every active query, each of
which answers with a fresh snapshot.

**What is on screen while that happens is your last-known-good data, not a
skeleton.** The replacement cache is seeded from the one it retires, so `data`
keeps its previous value and `loading` stays `false` across the gap; the first
snapshot on the new stream replaces the stale rows. There is nothing to opt
into:

```tsx
const { data, loading } = useSubscription('app', 'notes.list', {})
if (loading) return <Skeleton/>   // does NOT fire on a reconnect
return <NoteList notes={data}/>
```

Use [`useConnectionStatus`](/docs/ui/client-utilities/use-connection-status) if
you want to tell the user the rows may be a few seconds old — the data itself
never disappears from under them.

Three things are deliberately NOT carried across:

- **Optimistic patches.** They are client-local and belong to mutations that
  died with the old connection, so nothing could ever retract them. They are
  reverted when their mutation settles.
- **A cold-start error.** The new connection re-establishes the truth.
- **Entries nothing re-subscribes to.** A screen that unmounted during the
  reconnect does not pin its rows; the seed evicts on the normal inactive TTL.

### An auth change still blanks — on purpose

When the rebuild happens because the connection's *subject* changed — a cookie
login, a logout, a tenant switch, i.e. `useReconnect()` — **nothing** is carried
over and the screen does go back to its loading state.

That is not a gap in the feature, it is the point of the gate. The next subject
may be entitled to strictly less than the previous one, so painting the previous
subject's rows into their session, even for the moment before the first snapshot
lands, would be a data exposure. The same rule applies to
[`useRefreshSubscriptions`](/docs/ui/client-utilities/use-refresh-subscriptions),
which clears each entry's data on the same-socket re-auth path.

The short version: **a dropped connection keeps your screen, a change of
identity clears it.**

## Tenant Scoping

Tables with the `tenant()` mixin are scoped by the runtime using `ctx.request.subject.tenantId`. Do not add duplicate tenant predicates in query executors unless you are deliberately narrowing further.

## Inspect

The devtools subscription surfaces show active subscribers, recent deltas, and cache state. Use them when a query updates too often or not at all.

## Cost — how large may a live query be?

Every change re-runs the query and diffs the WHOLE result against the previous
one, so the cost is linear in the RESULT SIZE, not in the size of the change.
Measured on `diffRows`:

| result rows | one column changed | every row replaced |
| --- | --- | --- |
| 50 | 45 µs | — |
| 500 | 480 µs | — |
| 2 000 | 1.23 ms | 1.29 ms |
| 5 000 | 3.1 ms | — |

Two things follow, and the second is the one that surprises people:

- **The curve is linear, not quadratic.** Per-row cost is flat across a 100×
  growth (910 ns → 625 ns), so a large result gets slower in proportion and
  never falls off a cliff.
- **A one-column edit costs the same as replacing everything.** 2 000 rows with
  a single change is 1.23 ms; the same 2 000 rows entirely replaced is 1.29 ms —
  5 % more. The cost is the WALK, not the delta. Making your mutation smaller
  does not make the subscription cheaper.

So the number to design against is the RESULT SIZE. A few hundred rows is free.
A 5 000-row live query costs 3.1 ms of CPU per change, per replica — fine for a
dashboard that changes a few times a minute, wrong for one fed by a high-rate
writer. Page the query, or narrow it with a predicate, rather than reaching for
a bigger machine.

These numbers are asserted by `rowPatch.perf.test.ts`, so they are current
rather than a note somebody wrote down once.

## Fan-out — how many subscribers may one change wake?

A change wakes every subscription that reads the changed table, and the framework
already collapses the work they share: one READ per distinct query, one DIFF per
distinct `(query, base)`, one no-op comparison per distinct `(query, base)`. Fifty
screens on one query cost one of each, not fifty.

What does NOT collapse is what is genuinely per subscriber: re-running the query's
`guards:` and re-resolving row-level visibility. Those are re-run for every
subscriber on every delivery, on purpose — a role revoked or a share withdrawn has
to end the stream on the very NEXT delivery, not whenever a cache happens to
expire — and each of them can be a database round-trip.

So deliveries run **concurrently, up to a bound**. The default is 8 in flight.
Measured with 50 subscribers behind a 5 ms guard: 517 ms to serve all of them
serially, 72 ms at 8 lanes.

Declare it in `app.config.ts`, or override per deployment with the env var —
the env wins, because an operator acting on a running system outranks what the
project declared:

```ts
export default {
  type: 'api' as const, name: 'api',
  reactive: { deliveryConcurrency: 16, rawReadTrackingLimit: 64 },
}
```

```bash
VOLTRO_REACTIVE_DELIVERY_CONCURRENCY=16 voltro serve
```

Raise it when your guards or row filters hit the database and you have pool
headroom; set it to `1` for strictly one-at-a-time delivery. A value that is not
a positive integer is ignored rather than honoured — a concurrency of `0` is a
fan-out that delivers to nobody, and that is reachable through a typo in a values
file. Unbounded is deliberately not an option: one round-trip per subscriber at
the same instant starves the connection pool the request path shares, which is
slower than serial.

Per-subscriber ordering is unaffected: a change touches each subscription exactly
once. Order BETWEEN subscribers was never guaranteed.

### How many subscribers fit on one node?

There is a number, it is not a constant, and which number you get depends on a
property of your **queries** rather than of your scale. Re-derive it on your own
hardware with `node packages/runtime/scripts/fanout-ceiling.mjs`; the figures
below are the spread across three runs on a busy developer machine at 10 matched
writes per second, against a budget of 100 ms of event-loop time per second (10%
of one core).

| Subscriber population | Marginal CPU per subscriber | Subscribers per node |
| --- | --- | --- |
| **Shared** — N clients on the SAME query (a leaderboard, a shared board) | 0.5–0.9 µs | ≈ 11 000–20 000 |
| **Distinct** — N clients each on their OWN query (`where userId = me`) | 22–29 µs | ≈ 350–450 |

Ranges rather than single numbers, deliberately: that is the spread three runs
produced, and a ceiling quoted to three significant figures from one run is a
number somebody will hold you to.

The shared case is cheap because the memoisation above applies: one read, one
diff, N emits. The distinct case gets no sharing at all — the read, the diff and
the emit are all per subscriber — and a per-user dashboard is exactly that shape.
**Plan against the distinct number**, and note that it scales inversely with your
write rate: at 1 matched write per second it is ten times higher.

Two things that are easy to assume and are not true:

- **A more selective `where` buys no headroom.** Measured: 200 of 200
  subscribers whose predicate matched *nothing* were still woken by one write on
  their table. Every subscription is a dependent of its own table, so a change
  wakes all of them and each re-queries. The ceiling counts subscribers **on the
  table**, not subscribers whose predicate matches.
- **It is not 512.** That constant bounds `onChange` LISTENERS — one per declared
  subscription file, reaction or aggregate, bound once at boot. Every client
  subscription in a process shares the dispatcher's single listener, so ten
  thousand of them move that count by zero.

Past the ceiling the lever is horizontal: more nodes, each carrying fewer
subscribers. Change fan-out is already fleet-wide on postgres (LISTEN/NOTIFY) and
mysql/mariadb (binlog), so a second node needs no extra wiring — the cost being
budgeted here is the matcher and re-query CPU each node spends on ITS OWN
clients.

## Raw WebSocket gateways — `defineWebSocket`

Everything above rides the framework's subscription protocol, and that stays the answer for app realtime — live queries, optimistic patches, reconnect. A **gateway** exists for the other case: a FOREIGN protocol that needs a socket the framework does not speak — a Yjs provider, a legacy device fleet, an MQTT-over-WS bridge. It mounts its own upgrade path beside the rpc socket, in a `*.ws.ts` file discovered on **both** boot paths:

```ts
// gateways/yjs.ws.ts
import { defineWebSocket } from '@voltro/protocol'

export default defineWebSocket({
  path: '/gateways/yjs',
  auth: 'subject',   // REQUIRED, no default: 'subject' | 'public'
  onConnection: ({ send, close, onMessage, subject, headers, path }) => {
    const doc = attachDoc(subject!.id)
    onMessage((data) => doc.applyUpdate(data))     // binary-safe frames
    const stop = doc.onUpdate((update) => send(update))
    return () => { stop(); doc.release() }          // teardown
  },
})
```

The contract, in the order it protects you:

- **`auth` is mandatory and has no default.** `'subject'` runs the SAME auth chain as rpc/SSR *before* the upgrade — an unauthenticated caller gets `401` while the request is still plain http, and the connection is bound to the credential's expiry: when it lapses, the socket closes with application code `4001`, so a foreign client can re-auth and reconnect. `'public'` is a deliberate, written-down decision (a device fleet with protocol-level auth of its own).
- **Every gateway path is origin-checked at upgrade** — cross-origin means `403`, which closes cross-site WebSocket hijacking for your protocol exactly as for the framework's socket.
- **`onConnection({ send, close, onMessage, subject, headers, path })`** may return a teardown function — it runs on client disconnect, on credential expiry, and on server shutdown, so whatever the handler opened cannot outlive the socket.
- A plain GET on a gateway path answers `426 Upgrade Required`; two gateways declaring one path refuse the boot.

**The boundary to keep:** if your own UI needs live data, that is a query + `useSubscription`, never a gateway. A gateway hands you raw frames and none of the subscription protocol's guarantees — reach for it only when the CLIENT dictates the protocol.

## See also

- [Subscribers (`*.subscribe.ts`)](/docs/data/subscribers) — server-side, best-effort post-commit reactivity to a table (NOT the client hook on this page).
- [Streams](/docs/data/streams) — transient, non-database element feeds (no snapshot/reconnect replay).



---

<!-- source: en/data/streams.md -->
## Streams

_`*.stream.ts` + `*.stream.server.ts` pairs — one-shot server-to-client element streams with defineStream and useAgentStream._

A **stream** is a one-shot server-to-client feed. It emits plain elements, in order, and then finishes. It is not reactive, does not have snapshots or patches, and does not participate in auto-optimistic cache updates.

Use streams for transient work: LLM token events, progress updates, import logs, provider events, or any server-side operation where the client should see elements as they arrive.

Live — click Start and `{ n }` elements arrive one at a time (150ms apart), then
the stream finishes (one-shot push, no snapshot/delta):

```tsx
const s = useAgentStream('app', 'progress.count')
s.start({ to: 20 })   // s.events appends as elements arrive
```

## Minimal stream pair

Descriptor:

```ts
// apps/api/streams/ticker.stream.ts
import { defineStream } from '@voltro/protocol'
import { Schema } from 'effect'

export const ticker = defineStream({
  name:    'ticker.watch',
  openAccess: 'public market prices from an upstream feed — reads no table and '
    + 'nothing derived from the caller',
  input:   Schema.Struct({ symbol: Schema.String }),
  element: Schema.Struct({
    price: Schema.Number,
    at:    Schema.Date,
  }),
})
```

> **The access decision is what makes this file boot.** A wire-exposed stream
> must declare exactly one of `guards:`, `openAccess: '<reason>'` or
> `internal: true`, or the boot refuses it, naming the file. This one is
> genuinely open, so it says so — and the reason is the point: `openAccess`
> takes a sentence, not a boolean, because a reviewer has to be able to check
> the claim. A stream carrying *your* rows (an import log, an export feed) wants
> `guards:` instead. Full rules:
> [Authorization](/docs/authentication/authorization#every-procedure-decides-guards-or-openaccess).

Server executor:

```ts
// apps/api/streams/ticker.stream.server.ts
import { Stream } from 'effect'

export default (input: { symbol: string }) =>
  Stream.fromAsyncIterable(
    externalTickerStream(input.symbol),
    (error) => error,
  )
```

The executor can return `Stream`, `Effect<Stream>`, or a `Promise<Stream>`. The runtime binds it with `bindStream` and interrupts it when the client cancels or disconnects.

## Client hook

`useAgentStream` consumes any `defineStream` RPC, not only AI agents:

```tsx
import { useAgentStream } from '@voltro/client'

const ticker = useAgentStream<{ price: number; at: Date }>('app', 'ticker.watch')

return (
  <>
    <button onClick={() => ticker.start({ symbol: 'BTC' })}>Start</button>
    <button onClick={ticker.cancel}>Cancel</button>
    <ul>
      {ticker.events.map((event, index) => (
        <li key={index}>{event.price}</li>
      ))}
    </ul>
  </>
)
```

State shape:

| Field | Meaning |
|---|---|
| `events` | Elements received so far, in order. |
| `status` | `'idle'`, `'streaming'`, `'done'`, or `'error'`. |
| `error` | Last stream failure when `status === 'error'`. |
| `start(input?)` | Starts a new stream and clears previous events. |
| `cancel()` | Interrupts the in-flight stream. |

## Stream vs query

| | Query | Stream |
|---|---|---|
| Files | `*.query.ts` + `*.query.server.ts` | `*.stream.ts` + `*.stream.server.ts` |
| Descriptor field | `output` | `element` |
| Client hook | `useSubscription` | `useAgentStream` |
| Wire element | Subscription event envelope | Plain element |
| Reactivity | Re-runs on matching writes | No automatic reactivity |
| Persistence | Usually backed by database state | Transient unless you persist yourself |
| Best for | Live data views | Token/progress/event feeds |

If users should be able to reload the page and still see the result, persist rows and expose them through a query. If the stream is just live progress for the current run, keep it as a stream.

## AI streams

Transient agent runs are just streams whose elements are token/tool events. `useAgent` is an ergonomic wrapper over `useAgentStream`; durable chat uses a different pattern: an action writes/patches `agent_messages`, and a query streams those persisted rows.

See [AI streaming](/docs/ai/streaming) and [Agents](/docs/ai/agents) for those patterns.

## Anti-patterns

- **Using streams as a database subscription replacement.** Queries already do this and handle reconnect snapshots.
- **Streaming data that must be durable.** Put it in a table and subscribe to a query.
- **Forgetting cancellation.** Long streams should release upstream resources when interrupted.

## See also

- [Subscriptions](/docs/data/subscriptions) — live, reactive query data with snapshot + reconnect replay (use this for durable row sets).
- [Subscribers (`*.subscribe.ts`)](/docs/data/subscribers) — server-side post-commit reactivity to a table.



---

<!-- source: en/data/events.md -->
## Events

_`*.event.ts` — ephemeral fan-out to connected clients with defineEvent, ctx.events.publish and useEvent. At-most-once, live, and it tells you when it lost something._

An **event** is a thing that *happened*. It has a time and no value afterwards — where a table row has a value and no time.

That distinction decides which primitive you want, and it is the only decision here that is hard to reverse later:

| You are modelling | Use | Because |
| --- | --- | --- |
| **what happened** — a game started, a door opened, a terminal confirmed a payment | **Events** (this page) | nothing to store; a late arrival wants what happens *next*, not the history |
| **what is** — the current roster, an order's status, a document | [Queries](/docs/data/queries) + [Subscriptions](/docs/data/subscriptions) | a late arrival wants the current value immediately |
| **what must happen, even if we crash** — charge a card, send an invoice | [Outbox](/docs/data/outbox) | needs persistence, retries and a delivery guarantee |

If you find yourself writing a table so that a subscriber fires, you want an event.

## Declare it

```ts
// events/gameLifecycle.event.ts
import { defineEvent } from '@voltro/protocol'
import { Schema } from 'effect'

export const gameStarted = defineEvent({
  name: 'games.started',
  key: Schema.Struct({ arenaId: Schema.String }),
  payload: Schema.Struct({
    gameId: Schema.String,
    gameType: Schema.Literal('evo5', 'evo6'),
    startedAt: Schema.Number,
  }),
  guards: [{ scope: 'display:read' }],
})
```

A `*.event.ts` file is **browser-safe** and may hold several declarations — a lifecycle's stages are one concept. Client and server import the *same* value, which is what makes the key and payload types identical at both ends.

**`key` is the address, and only the address.** A subscriber receives events published under a key it asked for, so the server never sends the others at all. Put in it what *routes* (`arenaId`) and nothing else — a discriminator your handler reads (`gameType`) is payload. Every key field fragments the subscriber set.

**`guards` decide who may listen**, in the same vocabulary a query uses, and they are checked *before* the subscription is registered — a refused client never holds one. The routing key is the guard input, so a resource-scoped guard (`{ scope: 'arena:read', from: 'arenaId' }`) sees which arena was asked for. A refusal reaches the client as the `ScopeError` the rpc declares.

**`openAccess: '<reason>'` is the other answer to the same question**, exactly as on a procedure. Under `security.defaultDeny` (the default) an event that declares *neither* `guards:` nor `openAccess:` refuses the boot, by name — an access decision nobody made is a hole, not a default. For an event that really is open — a public scoreboard, a status pulse — declare it and say why; the reason is what a reviewer reads and what `voltro doctor` prints beside the tag. Do not reach for a scope every caller already holds just to satisfy the gate: that guard reads as protection and enforces nothing.

```ts
export const scoreboardTick = defineEvent({
  name: 'scoreboard.tick',
  key: Schema.Struct({ arenaId: Schema.String }),
  payload: Schema.Struct({ score: Schema.Number }),
  openAccess: 'public scoreboard — carries no caller data',
})
```

**The tenant is not part of the key and must never be added.** It comes from the subject on both sides, so a cross-tenant delivery is impossible by construction rather than by remembering to filter.

The name shares the rpc tag space. Two declarations answering to one name fail the **boot**, not the first delivery.

## Publish it

Only the server publishes. A client-originated event is an [action](/docs/data/actions) that publishes — which means every publish has already passed a guard-checked, typed handler, and there is no "who may write to this channel" question to answer.

```ts
// actions/reportGameEvent.action.server.ts
export default (input, ctx) => Effect.gen(function* () {
  yield* ctx.events.publish(gameStarted, { arenaId: input.arenaId }, {
    gameId: input.gameId,
    gameType: 'evo5',
    startedAt: Date.now(),
  })
})
```

`publish` works from anywhere with a `ctx`: an action, a mutation, a workflow, a subscriber, a cron, a startup hook.

**Inside a mutation it fires on COMMIT, and not at all on rollback.** That is not a nicety: a display reacting to a game start the database rolled back happens on every constraint violation, every deadlock retry, every guard that fails *after* the publish line. Outside a transaction it fires immediately.

Three typed errors reach the **producer**, so a mismatch is one failing call rather than every deployment's handler breaking on a field that is not there: `EventPayloadInvalid`, `EventKeyInvalid`, `EventPayloadTooLarge`.

`EventPayloadTooLarge` fires at **7,500 bytes** for the encoded envelope (event
name + key + payload, as JSON). The ceiling is not arbitrary and it is not a
transport limit to tune around: an event says that something *happened*, so
`photo.added` carries a photo REFERENCE and the consumer fetches the photo
through a route that can stream, cache and authorize it. A payload approaching
this size is usually a read that has been pushed into a notification.

<Callout>
**Both handler styles publish.** `ctx.events.publish` returns an Effect, so the
`Effect.gen` form above is the idiomatic one — but `await ctx.events.publish(…)`
in a plain `async (input, ctx) => { … }` handler works too and resolves with the
same result. It used to hand back an unrun Effect: nothing published, nothing
errored, and the handler returned success.
</Callout>

## Consume it

```tsx
const { missed, status } = useEvent(gameStarted, arenaId ? { arenaId } : null, (payload) => {
  scene.switchTo('running', payload.gameId)
})
```

`payload` is typed from the descriptor — a wrong field name is a `tsc` error at this call site.

Everything you would otherwise hand-roll is gone, and each of these was a real bug in apps that built this on a reactive list:

- **No history on mount.** A fresh subscriber gets what happens *from now on*. No `seen` set, no `initialized` flag, no window.
- **Exactly once, even under React `StrictMode`** — where every effect runs twice and a naive subscription fires each handler twice, in development only.
- **A changing handler does not resubscribe.** Every call site passes an inline arrow; putting it in a dependency array rebuilds the subscription on every render and loses whatever arrives in the gap.
- **A key change is a clean switch** — the old subscription ends before the new one starts.
- **`key: null` means "not yet"**: no subscription, `status: 'idle'`. You never need a placeholder key.

## Showing that it is showing stale

`useEvent` returns `{ status, missed, lastMiss }` — `status` is
`'idle' | 'connecting' | 'live'`, so `status === 'live'` is your connected flag
and needs no extra plumbing.

```tsx
const { status, missed } = useEvent(gameStarted, { arenaId }, onStart)

// A screen nobody is standing at should say when it stopped being current.
{status !== 'live' && <Badge>reconnecting…</Badge>}
{missed > 0 && <Badge>{missed} missed — refreshing</Badge>}
```

This matters most where nobody is watching the tab. A wall display that loses
its connection keeps rendering the last thing it received, and from across the
room "stale" and "current" look identical. The difference between *showing old
data* and *showing that it is showing old data* is one badge.

`missed` is the other half: it is COMPUTED, never estimated, so a non-zero value
means deliveries provably did not arrive — worth surfacing rather than hiding,
because the refresh that follows is visible anyway.


## What it guarantees — read this before you build on it

- **At-most-once, best-effort, live.** No persistence, no retry, no redelivery. For guarantees use the [outbox](/docs/data/outbox); this is the other axis.
- **Ordered per publishing instance per key.** *Not* globally per key — two instances publishing the same key have no shared counter, and we do not promise an order we cannot keep.
- **Guards are re-checked on EVERY delivery**, exactly as a live query's are. A resource-scoped guard (`{ scope: 'arena:read', from: 'arenaId' }`) runs its resolver each time, so un-sharing a resource or ending a membership stops the stream at the next delivery — the client is told, not silently skipped. What this does *not* catch is a ROLE revoked on the subject itself: those scopes were captured when the subscription opened. That half is covered by the credential bound below.
- **A subscription cannot outlive the credential that authorized it.** When the session carries an expiry, the stream ends at it — and `useEvent` reconnects immediately, which is a NEW request, so the subject is resolved afresh and the guards run again for real. Still entitled: it continues and your app sees nothing. No longer entitled: the reconnect is refused, loudly. You write no reconnect handling for this; it is the existing retry doing its job. Note the limit precisely — this bounds EXPIRY, not revocation.
- **Payloads are capped at 7,500 bytes** of encoded envelope, on **every** dialect. An event says something happened, so carry a photo *reference*, not a photo.

### `missed` is a number, not a feeling

When deliveries are lost, you are told **how many** and **why**:

```tsx
useEvent(gameStarted, key, handler, {
  onMissed: ({ count, reason }) => resyncFromServer(count),   // reason: 'buffer' | 'resume'
})
```

Every delivery carries a serial, and the server keeps the highest it has seen — so a loss is arithmetic (what you were owed, minus what could be replayed), never an estimate. `buffer` means your client fell behind and the oldest were dropped; `resume` means a reconnect asked for messages older than the server still holds.

This matters more than it sounds. Silence is the one outcome nothing can be built on: a display cannot tell "no game started" from "I missed the start signal".

### Reconnects resume; mounts do not

These read as one contradiction — *never replay history* against *never lose a message* — and they are two different questions:

- **A first attach** starts empty. Set `rewind: true` on the descriptor if you want the recent buffer instead.
- **A reconnect** continues from the last serial that subscription saw. `useEvent` does this for you, including after a deploy or a proxy timeout.

The buffer is deliberately small — tens of messages, minutes. Anything larger is a durable queue, and the framework already has one.

## Triggering a workflow from an event

```ts
triggerWorkflow({ on: gameStarted, workflow: 'postGameReport' })
```

`on:` takes the descriptor and reads its name, so renaming the event moves this call site with it. The older string form (`event: 'games.started'`) still works and is going away: with a string, a rename leaves the trigger matching nothing and the workflow simply never runs again — nothing errors, which is the same silence this whole primitive exists to remove.

## Reaching HTTP receivers too

An event can also be delivered to subscribed webhook targets — the third audience of the same declaration:

```ts
export const orderPaid = defineEvent({
  name: 'orders.paid',
  key: Schema.Struct({ orderId: Schema.String }),
  payload: Schema.Struct({ total: Schema.Number }),
  webhook: { description: 'An order was paid', version: 2 },
})
```

One `publish` now reaches connected clients, every matching workflow trigger, and every subscribed HTTP target. Without this an app that does both declares the thing twice, in two shapes, and the two drift.

The `webhook:` block is namespaced because its settings mean nothing to the other audiences — a top-level `rateLimit` would read as if it throttled client delivery, which it does not — it is a ceiling on webhook deliveries only. Requires [`@voltro/plugin-webhooks`](/docs/plugins/webhooks); absent, it costs nothing.

## Across instances

Local delivery always works. For fan-out across replicas the event rides postgres `LISTEN/NOTIFY` or [`@voltro/plugin-broadcast`](/docs/plugins/broadcast) (Redis / NATS), exactly like change events — and a broker outage degrades cross-replica delivery without touching local subscribers.

**Each declared event gets its own channel** (`<namespace>:events:<name>`, where
the namespace defaults to your app's name — see
[broadcast](/docs/plugins/broadcast)), and a
replica subscribes to it only while it actually has a local subscriber. This
matters as soon as one event is much busier than the others: on a single shared
channel every replica receives, decodes and tracks every event of every peer,
including the ones it serves no clients for. With five replicas and a high-rate
event whose subscribers all sit on one of them, four were doing that work and
discarding the result.

Nothing to configure — it follows from the declaration. The one operational
consequence: during a **rolling deploy** replicas on different framework versions
use different channel names, so cross-replica delivery is degraded for the length
of the rollout. Local delivery on each replica is unaffected throughout.

## Measured against socket.io

Same machine, same Redis, same topology, back to back:

| | p50 (median of 5) | p50 range | p99 (median of 5) | p99 range |
| --- | --- | --- | --- | --- |
| **Voltro**, publisher → Redis → subscriber | **0.68 ms** | 0.58–0.86 | 7.01 ms | 4.50–13.36 |
| **socket.io + @socket.io/redis-adapter** | 1.31 ms | 1.16–1.73 | **4.52 ms** | 4.33–7.83 |

Five runs each, alternating, on one machine. **The p50 ranges do not overlap —
that is a real ~2x advantage.** The p99 ranges DO overlap, so the tail difference
is weaker evidence than the medians.

**An earlier version of this table reported one run each** and claimed "32%
faster at the median, 44% worse at the tail". Both numbers were noise: the
median advantage is nearer 2x and the tail gap is inside the overlap. A single
measurement presented as a fact is the defect this framework spends its time
removing, and it was committed here.

**Where our tail comes from, measured rather than guessed.** Splitting the
publish path: our own code — building the envelope, the Effect fiber, handing
off — costs **p50 0.056 ms / p99 0.444 ms**. Waiting for Redis to acknowledge
costs **p50 1.17 ms / p99 6.43 ms**. So roughly 0.4 ms of a 7 ms tail is ours;
the rest is the broker round-trip, which socket.io pays too. There is no
code-level tail defect to fix here — on this machine the number is dominated by
Docker's network stack.

The script is in the repo (`scripts/bench/socketio-cross-replica.mjs`) so the
number can be re-taken rather than believed. It is not a test: keeping a
competitor in the dependency tree to hold a number green is the wrong trade.

**Topology is what makes this a comparison at all.** Two server instances share
one Redis; the client hangs off instance B and every emit is issued on instance
A. The first version measured socket.io on a plain localhost websocket and came
out 3x faster — which proved nothing, because that is one hop and this is two
through a broker.

Hosted products (Firebase, Pusher, Ably) are deliberately absent. Measuring them
honestly needs their accounts, regions and tiers, and a wrong number about
someone else's product is worse than no number.


## The hard questions, and our answers

| the question | the answer here |
| --- | --- |
| Does a client learn that deliveries were missed? | Yes — `missed` is COMPUTED from per-origin serials against a watermark, never estimated |
| Can a late arrival tell "nothing happened" from "I was not listening"? | Yes — every delivery carries `prior`, the watermark before it was accepted |
| Is a subscription authorized, or only the connection? | Per subscription, on the routing key, re-checked per delivery |
| Does a subscription outlive the credential that authorized it? | No — bounded by the credential's verified expiry, cookie AND bearer |
| Does the link heal itself after a broker outage? | Yes — no restart, no app-side retry, no resubscribe |
| Does a degraded network lose messages or only slow them? | Only slows them — 7x the median latency, zero loss |
| Does fan-out cost grow with subscriber count? | No — 0.027 µs per delivery, flat from 1 to 100 |
| Are channels typed, or strings? | Typed — a rename is a compile error |
| Is a declared event nothing publishes reported? | Yes, at boot, reading sibling apps in the workspace |
| Is cross-replica fan-out separated per app by default? | Yes — the namespace derives from the app name |

**Every row is enforced by a test** (`realtimeProperties.test.ts`) that fails if
the proof behind it disappears. A property may not be claimed here without
something in the repository that demonstrates it.

**Why this is not a benchmark against other products.** A table of our measured
numbers beside someone else's published ones is not a comparison — it is two
things in a row. Benchmarking a hosted competitor honestly needs their accounts,
regions, tiers and retry policies, and a wrong number about someone else's
product is worse than no number. What decides a choice anyway is not the
microseconds; it is whether the system answers these questions at all. Each
answer above is checkable against this repository by anyone, which is the
opposite of a claim.


## What you can build — and what to reach for

| what you want to build | reach for |
| --- | --- |
| a list that updates as rows change | `useSubscription` |
| a huge list without loading all of it | `useWindowedSubscription` |
| an edit that appears before the server confirms | `useMutation` (optimistic is derived) |
| a signal with no row behind it — a game start, a trigger | `defineEvent` |
| a 60 Hz value where only the newest matters | `defineEvent` + `delivery: 'latest'` |
| knowing a delivery was provably missed | `useEvent` → `missed` |
| who is online in a room | `usePresence` |
| who is typing right now | `useTyping` |
| showing that the screen went stale | `useEvent` → `status`, or `useConnectionStatus` |
| fan-out across replicas | `broadcastPlugin()` |
| durable work with progress a client can watch | `useWorkflow` |
| an in-app inbox | `useInbox` |
| delivering an event to a third party | `@voltro/plugin-webhooks` |
| gating who may subscribe | `guards:` on the event |
| an upload whose progress the UI follows | `useUpload` |

This table is a TEST (`realtimeCapabilities.test.ts`), not a claim: each row
asserts its primitive is still exported, so a capability that loses its
primitive to a rename goes red in CI rather than being discovered by whoever
tries to build it.

**Three of these are the ones people usually reach for wrongly.** A value that
changes many times a second is an EVENT, not a row — writing it to a table wakes
every subscriber of every query reading that table, and each pays a full re-diff.
"Who is online" is presence rather than a table, because the answer is ephemeral
and per-connection. And "did anything get lost" has a real answer (`missed`),
computed rather than estimated, so you do not have to build a heartbeat of your
own to find out.


## The four costs side by side

Every number below is asserted by a test in the repo, not quoted from a report —
each surface has a `*.perf.test.ts` that prints what it measured.

| primitive | operation | cost | scales with |
| --- | --- | --- | --- |
| **Events** | `ctx.events.publish` | **4.3 µs** (~232 k/s) | nothing — flat |
| **Events** | delivery to a subscriber | **0.027 µs** | subscribers, linearly and cheaply |
| **Presence** | a heartbeat | **0.16 µs** | nothing — flat |
| **Presence** | a roster read, 10 k members | **547 µs** | the ROOM |
| **Records** | a live query re-diff, 5 000 rows | **3 062 µs** | the RESULT SET |
| **Broadcast** | cross-replica, real Redis | **p50 1.1 ms · p99 11.2 ms** | the network |

**The comparison is the useful part.** Publishing an event costs about a
thousandth of what re-diffing a large live query does, and that ratio — not
either number — is what should decide between them. A value that changes at 60 Hz
belongs in an event; the same value written to a table wakes every subscriber of
every query reading it, and each pays the full walk.

**Two of the four are flat and two are not.** A publish and a heartbeat cost the
same whatever the load, so they scale by adding replicas. A roster read is
linear in the room and a live-query diff is linear in the RESULT SET — including
when one column of one row changed, because the cost is the walk rather than the
patch. Those are the two numbers to keep an eye on as an app grows.

**Cross-replica adds milliseconds, not microseconds**, and that is a network
crossing rather than framework overhead. Under an injected 20 ms ± 10 jitter it
becomes p50 26 ms / p99 89 ms; adding a 50 KB/s ceiling makes it p50 188 ms — and
in every one of those runs, all 200 messages arrive. Degradation costs latency,
never messages.


## Throughput — the numbers, and where this is the wrong primitive

Measured on one core, publish path only:

| | |
| --- | --- |
| `bus.publish`, 1–100 subscribers | ~1.5µs (**~670,000/s**) |
| `bus.publish`, 1000 subscribers | ~3.1µs (~325,000/s) |
| `ctx.events.publish` (validation + size gate + bus) | ~4.1µs (**~240,000/s**) |

**Across replicas**, measured over a real Redis with two processes — 200 of 200
delivered, no loss:

| p50 | p95 | p99 | max |
| --- | --- | --- | --- |
| 1.67 ms | 2.91 ms | 6.58 ms | 9.95 ms |

That is the broker round trip plus both bus hops. It is the number that matters
for a display in another pod, and it is the one to compare against a hosted
realtime service — where the same hop is a network round trip to someone else's
region.

**Fan-out is nearly free.** One subscriber and a hundred cost the same — the
per-publish work dominates, not the delivery loop. What you pay per subscriber is
the wire encode on its own subscription, not anything in the bus.

For a game lifecycle — eight stage events per game, one publish each — that is
several orders of magnitude of headroom. Even 100 players at 60Hz (6,000
events/s) sits at ~2.5% of one core.

### Where it stops being the right tool

Not at a throughput number, but at a **semantic** one: this primitive guarantees
at-most-once delivery of *every* message, with gap accounting. For a 60Hz stream
of positions or cursors, that guarantee costs something and buys nothing —
**nobody needs frame 1 once frame 2 has arrived.** You want last-value-wins state,
not a delivery log.

### `delivery: 'latest'` — when only the current value matters

Declare it, and the framework stops treating a superseded value as a loss:

```ts
export default defineEvent({
  name: 'player.moved',
  key: Schema.Struct({ arenaId: Schema.String }),
  payload: Schema.Struct({ playerId: Schema.String, x: Schema.Number, y: Schema.Number }),
  access: 'authenticated',
  delivery: 'latest',
})
```

| | `each` (default) | `latest` |
| --- | --- | --- |
| A slow subscriber | keeps the newest, is told how many it lost | receives the current value |
| Gap reporting | `missed` is computed and delivered | none — nothing was lost |
| Server retention | up to 64 messages / 5 minutes | one value |
| Reconnect | replays what is retained, reports the shortfall | hands over the current value |

The distinction is **semantic, not performance**. `latest` is not "the fast mode":
it changes what a missing message *means*. Choosing it for a stream where each
delivery matters drops the ones in between; choosing `each` for a per-frame
stream makes a slow client work through a backlog to reach a state it could have
had immediately, and report a "loss" that was never one.

The test: **would a deployment be wrong to miss one?**

<Callout type="warn">
`delivery: 'latest'` cannot be combined with `webhook`, and the declaration is
refused. `latest` says a superseded delivery did not matter — but a webhook
delivery is a durable side effect at a third party, and one already sent cannot
be superseded. A 60Hz event with an HTTP audience is also 60 deliveries per
second *per subscribed target*, and the webhook rate limit **defers** the excess
as pending rows rather than dropping it, so the symptom is a growing table rather
than an error anyone would look at. Publish the high-rate event for clients and a
separate, coarser one for the outside world.
</Callout>

### Still worth avoiding

- **Payloads over a few hundred bytes at high rate.** The size gate stops you at
  7,500 bytes, and long before that the wire encode per subscriber becomes the
  cost. Send a reference.
- **Per-frame data as an event at all.** Even under `latest`, 60Hz of positions
  is 60 encodes per second per subscriber. Coalescing on the client and
  publishing at 10–20Hz is usually indistinguishable to a human and an order of
  magnitude cheaper.

The honest rule: **use an event when a deployment would be wrong to miss one.** If
missing one is fine because the next one supersedes it, either declare
`delivery: 'latest'` or model it as state — a table, or a value the client
replaces.

## Testing

```ts
import { testEventBus } from '@voltro/testing'

const events = testEventBus()
const display = events.subscribe(gameStarted, { arenaId: 'a1' })
await events.publish(gameStarted, { arenaId: 'a1' }, { gameId: 'g1', gameType: 'evo5', startedAt: 0 })
expect(display.received).toEqual([{ gameId: 'g1', gameType: 'evo5', startedAt: 0 }])

// Force a loss deterministically instead of racing a queue:
events.skipSerials(gameStarted, { arenaId: 'a1' }, 5)
```

It drives the real bus and the real publish path — validation, the size gate and serials all behave as they do in production — so a test cannot pass on a payload the server would reject.

## Evolving a payload

Clients decode against **their own** copy of the schema. Adding a field is safe. **Removing a required field breaks clients still running the old bundle**, loudly, at decode time — which is better than a silent `undefined` in a handler, and worth knowing if you ship to devices that do not reload for months. Treat those deployments as additive-only.

## Anti-pattern: events as rows

If you have this, replace it:

```tsx
// ✗ an events TABLE, reconstructed into "new" on the client
const { data } = useSubscription('app', 'realtime.list', { limit: 500 })
const seen = useRef(new Set()); const initialized = useRef(false)
useEffect(() => { /* mark everything seen on the first pass, then diff */ }, [data])
```

Three bugs in nine lines, and every consumer has to get all three right: the `seen` set, the `initialized` flag (without it, loading the page replays 500 old events into a live system), and `limit` (a silent ceiling — nothing tells you when more than 500 arrive between renders). The table also grows forever and holds rows nobody reads twice.

Migrating is mechanical: declare the event, replace the insert with `ctx.events.publish`, replace the hook with `useEvent`, and drop the table.

## What `defineEvent` refuses, and why

```ts
defineEvent({ name: 'orders paid', … })              // ✗ whitespace — see below
defineEvent({ name: 'orders.paid', guards: [], … })  // ✗ enforces nothing
defineEvent({ name: 'x', openAccess: '', … })        // ✗ the reason is the point
defineEvent({ name: 'x', openAccess: 'why', guards: [{ scope: 's' }], … })  // ✗ two decisions
defineEvent({ name: 'x', webhook: { rateLimit: { perMinute: 0 } } })  // ✗ never delivers
```

**Whitespace in a name is a broker-level failure, not a style rule.** The name
becomes a broker SUBJECT segment. NATS refuses a subject containing whitespace
and delivers nothing — with no error on the publishing side. An app that works
on Redis therefore stops working when the transport changes, silently and in
production only. Use a dot to namespace: `orders.paid`.

**`guards: []` is refused** because it reads at the call site as if the event
were protected and enforces nothing — the empty list never reaches the check.
An event with no decision at all does not boot under `security.defaultDeny`;
declare `openAccess: '<why>'` for a deliberately open one. **An empty
`openAccess` reason is refused** — the reason is the reviewable half of the
decision — and **`openAccess` + `guards` together are refused**: two decisions
say the event is protected AND open.

**`rateLimit: { perMinute: 0 }`** defers every delivery forever. There is no
"unlimited" spelling — omit `rateLimit` for no ceiling. **`version: 0`** would
make a subscriber pinned to 1 read the event as *behind*, the opposite of what a
bump is for.


## See also

[Subscriptions](/docs/data/subscriptions) · [Outbox](/docs/data/outbox) · [Subscribers](/docs/data/subscribers) · [Streams](/docs/data/streams)



---

<!-- source: en/data/rest-routes.md -->
## REST routes

_defineRestRoute — public raw-HTTP endpoints (URL + JSON) for third parties that don't speak the rpc WebSocket. Schema-typed input/output, guards, deprecation/sunset, mounted on the same listener._

The five primitives above ride one WebSocket — great for your own UI, wrong for a third party calling your API with a plain URL + JSON body. `defineRestRoute` (from `@voltro/protocol/rest`) is the public raw-HTTP surface: a schema-typed request/response endpoint that desugars to exactly one HTTP route on the **same** listener the rpc server + plugin routes use. It is NOT a separate runtime.

```tsx
// routes/v1/customers.list.route.tsx
import { defineRestRoute, requireScope } from '@voltro/protocol/rest'
import { Schema } from 'effect'

export default defineRestRoute({
  method: 'GET',
  path:   '/v1/customers',
  // Input shape is `{ query?, params?, body? }`. The desugar parses the
  // query string, path params, and JSON body into this, then decodes it.
  input:  Schema.Struct({
    query: Schema.Struct({
      limit:  Schema.optional(Schema.Number.pipe(Schema.between(1, 100))),
      cursor: Schema.optional(Schema.String),
    }),
  }),
  output: Schema.Struct({ data: Schema.Array(Customer), nextCursor: Schema.NullOr(Schema.String) }),
  summary: 'List customers',                  // OpenAPI metadata — inert until a generator consumes it
  guards:  [requireScope('customers:read')],  // → 403 when the subject lacks the scope
  handler: async ({ query }, ctx) => {
    const limit = query.limit ?? 20
    // ctx = { subject, headers, store }. The auth resolver populates
    // `subject`; `store` is the framework DataStore (cast at the call site).
    return { data, nextCursor }
  },
})
```

## Registration

REST routes are NOT auto-discovered by file suffix — register them explicitly via `restRoutes` in `app.config.ts` (the `*.route.tsx` filename is convention, not magic):

```ts
import listCustomers from './routes/v1/customers.list.route'

export default {
  type: 'api' as const,
  name: 'publicApi',
  restRoutes: [listCustomers],
}
```

The module's `default` export is the descriptor — one descriptor per file, the same one-procedure-per-file discipline as `*.query.ts` / `*.mutation.ts`.

## The request lifecycle (the desugar)

Each descriptor becomes one HTTP route. Per request, in order:

| Step | Outcome on failure |
|---|---|
| 1. Method gate | `405 Method Not Allowed` (`Allow:` header) |
| 2. Sunset gate (past `sunset` date) | `410 Gone` + replacement pointer |
| 3. Input decode (`{ query, params, body }` → schema) | `400 Bad Request` |
| 4. Guards (after decode + ctx resolution) | the guard's `{ status, message }` (e.g. `403`) |
| 5. `await handler(input, ctx)` | a thrown `{ status, message }` → that status; anything else → `500` |
| 6. Output encode → JSON | `200 OK` |

`deprecated` sets a `Deprecation: true` response header; `sunset` (an ISO date) sets a `Sunset:` header and, once past, flips the route to `410`. Both are pure metadata otherwise — the route keeps working until the sunset date.

## `ctx` — what the handler gets

```ts
interface RestRouteContext {
  readonly subject: Subject                              // resolved by the auth resolver
  readonly headers: Readonly<Record<string, string>>     // request headers
  readonly store: unknown                                // framework DataStore — cast at the call site
}
```

The same `AuthMiddleware`/`ConnectionInfoMiddleware` that gate the rpc surface run here too, so a forwarded session cookie resolves the same `Subject` + tenant as the WebSocket path.

## Guards

A `RestGuard` runs after input decode, before the handler. Return `undefined` to proceed, or `{ status, message }` to short-circuit:

```ts
import { requireScope, type RestGuard } from '@voltro/protocol/rest'

// Built-in: rejects 403 unless the subject carries the scope (admin '*' bypasses).
guards: [requireScope('customers:read')]

// Custom guard:
const requireApiKey: RestGuard = (ctx) =>
  ctx.subject.type === 'apiKey' ? undefined : { status: 401, message: 'API key required' }
```

## Methods — PATCH, HEAD and OPTIONS are first-class

`method:` accepts `GET`, `POST`, `PUT`, `PATCH`, `DELETE`, `HEAD` and `OPTIONS`, on REST routes and plugin HTTP routes alike. Two details:

- **HEAD is admitted wherever GET is** (RFC 9110): a `HEAD` request runs the GET route's whole pipeline — method gate, guards, handler — and the transport drops the body. You never declare a second route for it.
- **A wrong method is still a precise `405`**, with an `Allow:` header naming exactly the methods mounted on that path — including when several routes share one path.

## Body limits — `maxBodyBytes`

Every HTTP body read is capped at 8 MiB by default — `POST /rpc` (as it always was), plugin routes, REST routes and incoming webhooks. The app-wide cap is `http.maxBodyBytes` in `app.config.ts` (env override `VOLTRO_MAX_BODY_BYTES`); a route that legitimately takes more declares its own:

```ts
export default defineRestRoute({
  method: 'POST',
  path:   '/v1/import',
  maxBodyBytes: 64 * 1024 * 1024,   // this route only — the app cap stays 8 MiB
  // …
})
```

The same per-route override exists on an incoming webhook's handler (`maxBodyBytes`) — fat provider payloads are the normal case there, not the exception. Two details:

- **Routes that share one PATH share one body read.** The body is read once for the whole group, so the widest `maxBodyBytes` override in the group applies to the group.
- An oversized body answers `413` whether it announces itself (`Content-Length`) or arrives chunked — the counter cuts it at the cap and never buffers past it.

## Conditional GET — `etag: true`

```ts
export default defineRestRoute({
  method: 'GET',
  path:   '/v1/customers',
  etag:   true,   // GET only — ignored elsewhere
  // …
})
```

The route stamps a **weak, content-derived** `ETag` (`W/"<sha-1 of the encoded output>"`) on every `200`, and answers a matching `If-None-Match` with `304 Not Modified` — the tag, no body. Weak on purpose: the transport may vary the BYTES per content-encoding, but the representation is the same. (`voltro start` does the equivalent for the web app's HTML on its own — `If-None-Match` answers `304` for buffered `200`s, with weak `W/"md5"` tags over the *uncompressed* body; `no-store` responses excepted.)

## Idempotency (`Idempotency-Key`)

Set `idempotency: true` in `app.config.ts` and every mutating REST request (`POST`/`PUT`/`PATCH`/`DELETE`) that carries an `Idempotency-Key` header is deduplicated:

```ts
// app.config.ts
export default { type: 'api' as const, name: 'api', restRoutes: [chargeRoute], idempotency: true }
// or: idempotency: { header: 'Idempotency-Key', ttlMs: 86_400_000 }
```

```bash
# First call runs the handler + caches the response.
curl -XPOST https://api.example.com/v1/charge -H 'Idempotency-Key: pay-abc' -d '{"amount":99}'
#   → { "chargeId": "uuid-1", … }

# A retry with the SAME key replays the cached response — the handler does NOT run again.
curl -XPOST https://api.example.com/v1/charge -H 'Idempotency-Key: pay-abc' -d '{"amount":99}'
#   → { "chargeId": "uuid-1", … }   (+ response header `Idempotency-Replayed: true`)
```

- **Replay** within the window returns the first response verbatim, with `Idempotency-Replayed: true`.
- **In-flight duplicate** (the first call hasn't finished) gets `409`.
- Keys are scoped per `(tenant, method, path)`, persisted in `_voltro_idempotency`, expiring after `ttlMs` (default 24h).

This is the Stripe-style contract — the **client** opts in by sending the header. It's the standard for external API clients. Scope + guarantee:

- **REST/HTTP only.** It rides the `Idempotency-Key` HTTP header. The WS rpc transport (`useMutation` from your own frontend) has no per-call header — guard double-submit there with optimistic UI + a disabled button, not server idempotency.
- **Inbound webhooks already dedup** via [`@voltro/plugin-webhooks`](/docs/plugins/webhooks) (provider key + `_voltro_webhook_*`) — don't double-cover them.
- **Atomic claim, non-atomic completion.** Two concurrent same-key requests resolve to exactly one execution (the `UNIQUE(scope,key)` insert is the arbiter). But the cached response isn't committed in the handler's own transaction — a crash between the handler committing and the record flipping to `completed` leaves the key in-flight (a retry `409`s until the TTL lapses, then re-runs). REST handlers aren't auto-transactional, so this is the honest ceiling.

## API versions — opt-in `version:` + the sunset flow

A route that will evolve declares its version instead of baking it into the
path; `version: 'v2'` mounts under `/v2/…`:

```ts
// v2 — the current shape
export const listCustomers = defineRestRoute({
  method: 'GET',
  path: '/customers',
  version: 'v2',
  output: Schema.Struct({ data: Schema.Array(Customer), nextCursor: Schema.NullOr(Schema.String) }),
  handler: async (_i, ctx) => ({ data: await ctx.store.query(customers), nextCursor: null }),
})

// v1 — still mounted, deprecated, and gone on a date
export const listCustomersV1 = defineRestRoute({
  method: 'GET',
  path: '/customers',
  version: 'v1',
  deprecated: 'GET /v2/customers',   // Deprecation header + replacement pointer
  sunset: '2027-03-01',              // Sunset header; 410 Gone from this date
  output: Schema.Struct({ customers: Schema.Array(Customer) }),
  handler: async (_i, ctx) => ({ customers: await ctx.store.query(customers) }),
})
```

Two versions are **two descriptors** — the old one is ordinary code (visible,
testable, deletable), not an entry in a transformation DSL. While it lives,
responses carry `Deprecation: true` + `Sunset:`; past the date it answers
`410 Gone` with `{ version: 'v1', replacement: 'GET /v2/customers' }`. Then
you delete it. `version` is opt-in: a route without it keeps its literal path
(no auto-prefix), and declaring `version:` on a path that already starts with
`/vN/` is refused at definition — both spellings at once is never intended.
`publicApi` projections version the same way (`spec.version`, default `v1`),
and the OpenAPI doc groups each version's operations under a version tag with
`x-voltro-api-version` — one document, the `/vN/` paths already separate them.

**URI versioning only, on purpose.** Header- and media-type-versioning (the
NestJS options) are not supported: the OpenAPI document, cache keys and plain
`curl` are all path-shaped, and a version a URL cannot express is a version a
cached response cannot vary on. If an edge must accept `Accept-Version:`
headers, rewrite them to the path prefix at the proxy.

**And the rpc socket is deliberately outside this.** The generated client is
versioned with the server it was generated from — there is no `/v2` for
`useMutation`. Honest edge: a browser tab that stayed open across your deploy
runs the PREVIOUS client until reload; that skew window exists, it is small,
and URL versioning would not remove it.

## Projecting an existing procedure — `publicApi`

You often want to *offer* an API you don't consume from your own frontend. When the procedure already exists as a query / mutation / action, you don't need to rewrite it as a REST route — annotate it with `publicApi` and the framework mounts ONE HTTP route that runs the **same** handler, under the same guards:

```ts
// queries/absenceRequests.list.query.ts
export default defineQuery({
  name:   'absenceRequests.list',
  input:  Schema.Struct({ status: Schema.optional(Schema.String), limit: Schema.optional(Schema.Number) }),
  output: Schema.Array(AbsenceRequest),
  guards: [requireScope('absences:read')],
  publicApi: {},   // → GET /v1/absenceRequests/list?status=open&limit=20
})
```

- **Method** derives from the kind: query → `GET`, mutation / action → `POST` (override with `method`).
- **Path** derives from the tag: `/<version>/<tag-as-path>` (override with `path`; set `version`).
- **Input binding** follows the method: for `GET` the descriptor's `input` schema is bound to the **query string**, otherwise to the JSON **body**. So filter and pagination fields are plain URL params — no separate input shape.
- **Relations** need nothing extra: eager loading is resolved server-side by the executor, so `include` works identically over HTTP.
- **Authorization is the same code** as the WebSocket path — the declarative `guards:`, the row filter, and tenant scoping all run before the handler. A procedure that denies on the socket denies here.
- Also available per endpoint: `scopes` (extra API-key scopes), `rateLimit`, `idempotent`.
- **`idempotency:` covers these routes too**, on the same terms as a hand-written `restRoutes` entry above: one binding, one `Idempotency-Key` header, one `_voltro_idempotency` table, and identical behaviour under `voltro dev` and `voltro serve`. Projected routes and hand-written ones go through the same single projection, so it is not possible for one to deduplicate and the other not to.

This pairs with [`crud.list`](/docs/data/crud): `filter` / `paginate` / `sort` / `include` on the executor plus `publicApi: {}` on the descriptor is a complete, filtered, paginated REST list endpoint in one declaration.

Use `publicApi` when the procedure already exists and the derived URL is fine; use `defineRestRoute` when you need a hand-shaped URL, path params, or a response that isn't the procedure's output.

### Live updates over HTTP — `stream: 'sse'`

A third party that can't open your WebSocket can still follow changes: `stream: 'sse'` on a QUERY projects it as Server-Sent Events — the initial `snapshot`, then a `delta` per change, until the client disconnects.

```ts
export default defineQuery({
  name:   'orders.live',
  input:  Schema.Struct({ status: Schema.optional(Schema.String) }),
  output: Schema.Array(Order),
  guards: [requireScope('orders:read')],
  publicApi: { stream: 'sse' },   // → GET /v1/orders/live?status=open  (text/event-stream)
})
```

```js
// any EventSource client — no Voltro SDK needed
const es = new EventSource('/v1/orders/live?status=open')
es.addEventListener('snapshot', (e) => setRows(JSON.parse(e.data).data))
es.addEventListener('delta',    (e) => applyDelta(JSON.parse(e.data)))
```

Each event's `_tag` becomes the SSE `event:` name, so a client listens per kind instead of switching on a payload field. The framing handles the details that bite otherwise: embedded newlines are split across `data:` lines (a raw `\n` would truncate the event), a `retry:` hint is sent, and a keep-alive comment goes out every 15s so proxies don't drop an idle stream.

Same guarantees as the WebSocket path, because it is the same code: the declarative `guards:`, the row filter and tenant scoping all run before anything is emitted, and the client's disconnect tears the subscription down (including a disconnect *during* setup). A guard denial arrives as one `error` event rather than an HTTP status — by then the response headers are already sent.

`stream: 'sse'` on a mutation or action is ignored: there is nothing to subscribe to.

For a hand-written `defineRestRoute`, the same machinery is available directly — return `sse((emit) => unsubscribe)` from the handler and frame events with `sseFrame(event, data)` (both from `@voltro/protocol/rest`).

## Binary downloads — `bytes()`

A handler that serves a file, an export or any non-JSON body returns `bytes(stream, options)` — imported beside `defineRestRoute` / `sse`:

```ts
import { defineRestRoute, bytes, requireScope } from '@voltro/protocol/rest'

export default defineRestRoute({
  method: 'GET',
  path:   '/v1/exports/:id',
  guards: [requireScope('exports:read')],
  handler: async ({ params }, ctx) => {
    const file = await locateExport(params.id)
    // Lazy thunk form — the source is opened only when the response streams.
    return bytes(() => openExportStream(file), {
      contentType: 'application/zip',
      contentLength: file.size,
      contentDisposition: `attachment; filename="${file.name}"`,
    })
  },
})
```

- The first argument is a web `ReadableStream<Uint8Array>` — or the **lazy thunk form** `() => ReadableStream`, which defers opening the source until the response actually streams.
- The server **pipes without buffering** — a body larger than the heap is fine (the guarantee is exercised with a 256-MiB stream), and byte streams are **never compressed**.
- Everything before the handler still runs — method gate, sunset, input decode, guards — so a streaming route is exactly as gated as a buffered one.
- On a plugin HTTP route the same shape is `PluginHttpRouteResult.byteStream`.

### Idempotency × streams — decided

`streaming: true` on a method the idempotency binding claims (`POST`/`PUT`/`PATCH`/`DELETE`) is a **mount error**: a stream cannot cache a replayable body, so the idempotency claim could never complete — every retry would `409` until the TTL lapsed. The refusal names the two ways out: serve the stream on `GET`, or keep the idempotency binding away from the app's streaming routes. A handler that returns a stream *without* declaring `streaming: true` is caught at runtime instead — the claim is **released** so a retry re-processes.

## No multipart parser — a declared boundary

There is **no multipart parser** on REST or webhook routes — `multipart/form-data` against `/form/*` answers `415`, and a REST handler never sees parsed file parts. That boundary is deliberate, and this list of alternatives is complete:

- **File uploads** ride [`@voltro/plugin-storage`](/docs/plugins/storage)'s upload routes — a binary PUT plus a resumable, chunked upload with signed tickets. That is the sanctioned file path, not a workaround.
- **A provider that delivers webhooks as multipart** (the Mailgun-inbound class) needs, today, either a small parser proxy in front of the endpoint or the provider's JSON delivery mode where it offers one.

## REST route vs Action

Both are unary request/response. Pick by transport + audience:

| | [Action](/docs/data/actions) | REST route |
|---|---|---|
| Transport | rpc over WebSocket (+ `POST /rpc`) | public raw HTTP at a URL you choose |
| Caller | your own UI via `useAction` | third parties with `fetch` / curl / SDKs |
| Wire shape | the rpc JSON envelope | plain JSON body + status codes |
| Typed errors | `Schema.TaggedError` on the rpc channel | HTTP status codes (`throw { status, message }`) |

Use an action when your own client calls it; use a REST route when an external system needs a stable, documented URL. For signed inbound webhooks (Stripe, GitHub, …) use [`@voltro/plugin-webhooks`](/docs/plugins/webhooks) instead — it adds signature verification + idempotency on top.



---

<!-- source: en/data/aggregates.md -->
## Aggregates

_Pre-defined materialised queries that refresh on a schedule. The framework owns the lifecycle — discovery, refresh timer, cached rows, staleness metadata, dashboard surface._

> **Not what you want?** For an on-demand `count` / `sum` / `groupBy` / window that runs when called (not on a schedule), see [Aggregations](/docs/database/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

```ts
// 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

```ts
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](#why-no-virtual-table-integration) below).

### Read options

```ts
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.

```ts
// 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:

```ts
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

```ts
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:

```ts
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.

```ts
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:

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

```ts
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`](/docs/data/subscribers) |
| 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](/docs/plugins/analytics) |
| 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`](/docs/plugins/analytics). `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":

```ts
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`](/docs/data/reactions). 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`).

```ts
// 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:

```ts
{ 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:

```ts
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`:

```ts
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.

```ts
// 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)`:

```ts
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.

```ts
// 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`:

```ts
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.



---

<!-- source: en/data/crud.md -->
## CRUD helpers

_crud.* secure-default handler helpers — tenant-scoped reads, column redaction, and null-not-throw getById, so the CRUD tail of a handler is one honest line._

Most of a plain list / get / create / update / delete handler is the same five lines every time — and getting those lines subtly wrong is how data leaks. The `crud.*` helpers from `@voltro/runtime` give you the **executor** with the secure defaults baked in; you still write the descriptor (schemas + `guards`), which is where the browser-safe wire contract and the authorization live.

```ts
// accounts.list.query.server.ts
import { crud } from '@voltro/runtime'

export default crud.list('accounts', { redact: ['apiSecret'] })
```

```ts
// accounts.list.query.ts — the descriptor stays hand-written + browser-safe
import { defineQuery } from '@voltro/protocol'
import { Schema } from 'effect'

export default defineQuery({
  name: 'accounts.list',
  // the authorization lives here, not in the helper — crud.* adds no guard
  guards: [{ scope: 'accounts:read' }],
  input: Schema.Struct({}),
  // note: the wire output OMITS apiSecret, so it never reaches the client
  output: Schema.Array(Schema.Struct({ id: Schema.String, name: Schema.String })),
})
```

## What the defaults bake in

- **Tenant scope.** `crud.list` and `crud.getById` read through `ctx.store`, which auto-scopes a `tenant()` table. They never call `.unscoped()`, so a cross-tenant read is impossible through them — `payslips.list` cannot return another tenant's rows.
- **Redaction.** A column a generated read must never ship — a credential, a token hash, a salary — is stripped from every returned row (reads and the row a `create` / `update` echoes). Two sources: a column marked [`.serverOnly()`](/docs/database/sensitivity) is stripped **automatically** (declare the exposure policy once at the schema and every crud read respects it — the single-source form), plus the per-call `redact: [...]` option for anything not worth a schema marker. Declare the same omission in the descriptor's `output` schema so the column never reaches the client at all; the helper is the runtime guarantee that it doesn't, whatever the schema says.
- **No mass assignment of a `.serverOnly()` column.** The same marker is enforced on the **input** side: `crud.create` and `crud.update` **refuse** a payload that sets one, with `ServerOnlyColumnWrite` naming the offending columns, and nothing is written. `.serverOnly()` means *never crosses the wire*, so accepting it inbound is the same violation as leaking it, mirrored — a client could SET a column it is not allowed to READ. It is a refusal rather than a silent strip because a stripped field makes an attack look like a no-op and leaves an honest caller debugging a value that quietly did not land. A key present with the value `undefined` does not count as sent, so optional schema fields are unaffected. When the *server* legitimately needs to write one, do it from the handler with `ctx.store.insert` / `ctx.store.update` — those are unchanged; the refusal is on the generated path, which is the one fed straight from client input.
- **Keyed writes stay inside the caller's tenant.** `crud.update` and `crud.remove` address the row by `input.id`, which is client-supplied. On a [`tenant()`](/docs/multi-tenancy/mixin) table the store resolves that id inside `subject.tenantId`, so another tenant's id fails with `TenantRowNotFound` instead of writing. The error is the same whether the row is missing or foreign, on purpose — the pair would otherwise be a cross-tenant existence oracle.
- **`getById` returns `null`, never throws.** A reactive getter that throws takes its shared-WebSocket siblings down with it. `crud.getById` resolves `null` for an absent row.

## The helpers

| Helper | Executor it returns |
|---|---|
| `crud.list(table, { redact? })` | tenant-scoped list of every row, redacted |
| `crud.getById(table, { redact? })` | one row by `input.id`, or `null` — redacted |
| `crud.create(table, { redact? })` | insert `input`; id/tenant/audit auto-stamped; echoes the redacted row; refuses a `.serverOnly()` field in the input |
| `crud.update(table, { redact? })` | patch `{ id, ...patch }`; returns the updated row or `null`; tenant-resolved by id; refuses a `.serverOnly()` field in the patch |
| `crud.remove(table)` | delete `input.id`; returns `{ deleted }` |
| `crud.count(table, { filter? })` | `COUNT(*)` of the filtered, tenant-scoped set — the total for page-based UIs |

`redactColumns(rows, cols)` is exported standalone for a hand-written handler that isn't plain CRUD but still needs to redact declaratively.

## `crud.list` read ergonomics — filter, sort, paginate, include

A generated list isn't limited to "all rows". `crud.list` takes the ergonomics every real list view needs — all optional and additive on top of `redact`:

```ts
export default crud.list('absenceRequests', {
  filter:   (input) => ({ employeeId: input.employeeId, status: input.status }), // → WHERE
  paginate: true,                                        // ?page=3&pageSize=20  (or ?limit=20&offset=40)
  sort:     [{ column: 'createdAt', direction: 'desc' }], // multi-column
  include:  { employee: { with: { team: true } } },      // eager relations, nested filter/sort
  redact:   ['internalNote'],
})
```

- **`filter`** maps the request input to a `WHERE` — return a column→value map; an `undefined` field is ignored (an absent filter param is a no-op). Applied through the tenant-scoped `.where`.
- **`paginate: true`** accepts BOTH paging styles, so a caller uses whichever its UI thinks in: **`page`** (1-based) + **`pageSize`** (default 100), or **`limit`** / **`offset`** (defaults 100 / 0). `page` wins when both are sent, and a `page` below 1 clamps to the first page rather than producing a negative offset. For the total a page-based UI needs ("page 3 of 12"), pair it with `crud.count` — see below.
- **`maxPageSize`** caps how many rows ONE request may ask for (default **1000**). The page size is caller-controlled, so without a cap `?limit=1000000` is a one-request read of the whole table — and with `publicApi` that caller is anyone who can reach the URL. Values above the cap are clamped, not rejected; raise it deliberately for an export-style endpoint.
- **`sort`** is a multi-column `orderBy`, applied in order.
- **`include`** is the SAME spec [`.with(...)`](/docs/database/relations) takes, so nested relations and per-branch `where` / `orderBy` / `limit` (nested filtering and sort) all work. `getById` takes `include` too.


Pass `crud.count` the **same** `filter` as the list (share the option object) so the total and the pages can't disagree about which rows they mean. It ignores paging fields on the input — it counts the whole filtered set, not the current page:

```ts
// the total, for rendering "page 3 of 12" — SAME filter as the list
export default crud.count('absenceRequests', {
  filter: (input) => ({ employeeId: input.employeeId, status: input.status }),
})
```

Declare the filter / pagination fields in the descriptor's `input` schema so the client can pass them; the executor reads them off `input`.

**You don't need a `toView` projection layer** — the `output` schema already shapes the wire result. `crud.list` returns full rows, and on encode Effect **strips every column the output schema doesn't declare** (a tight `Schema.Struct({ id, name })` ships only `id` + `name`, whatever else the row holds), and a `timestampMs` field normalizes a `Date` to epoch ms. So select fields by naming them in `output`, and normalize dates with the wire field schemas — no per-table view function. (Renames are expressible via a Schema transform if you need them.)

### `columns` — don't even READ what you drop

The output schema stops a column reaching the client; `columns` stops it being read at all. Use it when a table carries something wide that a list view never shows — a long text body, a big `json()` blob:

```ts
export default crud.list('articles', { columns: ['id', 'title', 'createdAt'] })
// the large `body` is never SELECTed, transferred from the DB, or decoded
```

`.serverOnly()` columns are removed from the projection automatically — they're stripped from the response anyway, so reading them is pure waste.

**Trap:** an eager `include` branch joins on a foreign key, so a projection that omits that FK breaks the relation. Keep the FK in `columns` when you also pass `include`.

### `scope` — keep per-subject narrowing when you adopt the helper

`filter` builds the WHERE from the request; `scope` builds it from the **caller**:

```ts
crud.list('timeEntries', {
  filter: (input) => ({ status: input.status }),      // what the caller ASKED for
  scope:  (ctx)   => ({ ownerId: ctx.request.subject.id }),  // what it MAY SEE
})
```

A WHERE built only from `input` can express *the rows the caller asked for*, never
*the rows the caller may see*.

`scope` is merged **last**, so a request field of the same name can never widen it —
`?ownerId=someone-else` is simply overridden. That ordering is why the two are
separate options rather than one: only one of them is a security boundary, and kept
apart, *"does this list declare a `scope`?"* is a question a reviewer — or a future
boot audit — can actually ask. Folded into `filter`, it becomes *"does this filter
happen to read ctx somewhere in its body?"*, which nothing can check. Tenant scope still applies automatically; anything
narrower — owner, team, role — does not. So replacing a hand-written handler that
carried such a narrowing with `crud.list` **widens the result set, silently and
without an error**. One app lost exactly that across eight list views.

Use the same filter for `crud.count`, or the total contradicts the pages —
"showing 10 of 4000" on a page holding ten rows.

The reason this is an option rather than a reason to leave: hand-writing the query
to get the narrowing also forfeits `serverOnly` stripping and the page-size clamp.
A narrowing requirement should not cost you the safety rails.

## What they deliberately don't do — authorization

A guard runs *before* the executor, so gating lives on the **descriptor**, not the handler — an executor cannot gate itself. Keep every write descriptor guarded:

```ts
export default defineMutation({
  name: 'accounts.create',
  input: AccountInput,
  output: Account,
  guards: [requireScope('accounts:write')],   // ← the gate; crud.create does not add one
})
```

## Scope — why the schema is still hand-written

These helpers give you the secure **handler**, not schema derivation. Deriving the descriptor's `input`/`output` from the table automatically would need the table VALUE inside the descriptor file — and a descriptor is loaded value-level by the browser client, so importing a table there drags the store and driver into the browser bundle (the boundary guard aborts the boot; `rowSchema` is server-only for exactly this reason). Full schema-derivation, and a boot audit that fails when a `tenant()` table's list reads unscoped or a write goes ungated, are a separate planned pass — the handler defaults above are the part that ships browser-safe today and closes the leak class.



---

<!-- source: en/data/subscribers.md -->
## Subscribers

_Per-table post-commit reactivity via file convention. Default-exported defineSubscriber({ table, on, handler }) — fires AFTER commit, best-effort, fire-and-forget for async handlers._

Use a `*.subscribe.ts` file when you want code to **run after every commit** to a specific table — refresh a search index, emit an external notification, invalidate a cache, push to a worker queue. The file convention is parallel to `*.startup.ts` / `*.cron.tsx` / `*.webhook.tsx`: drop a file matching the suffix anywhere under `apps/<api>/`, default-export a `defineSubscriber({...})`, the framework discovers + binds it at boot.

Subscribers are deliberately **best-effort** + **non-durable**. For crash-safe async work — "a row changed, now run a workflow" — reach for a [reaction](/docs/data/reactions) instead.

## File shape

```ts
// apps/api/subscribers/auditUserChanges.subscribe.ts
import { defineSubscriber } from '@voltro/runtime'

export default defineSubscriber({
  table: 'users',
  on:    'any',          // 'insert' | 'update' | 'delete' | 'any' | ['insert','delete']
  handler: async (event, ctx) => {
    ctx.log.info('user changed', {
      op: event.op,
      id: event.new?.id ?? event.old?.id,
    })
    // event.new — present on insert + update; null on delete
    // event.old — present on update + delete; null on insert
  },
})
```

The default export must be a `defineSubscriber({...})` result. The file is identified by suffix (`*.subscribe.ts` / `*.subscribe.tsx`).

## `ctx.store` — reading and writing back

`ctx` carries `log`, `id`, and **`store`** — the same store surface a handler
uses, so a subscriber can act on the change it just saw:

```ts
export default defineSubscriber({
  table: 'orders',
  on:    'insert',
  handler: async (event, ctx) => {
    await ctx.store.insert('order_audit', {
      orderId:  String(event.new?.['id']),
      tenantId: String(event.new?.['tenantId']),   // explicit — see below
    })
  },
})
```

**`ctx.store` is NOT tenant-scoped.** A subscriber fires from the change stream,
not from a request, so there is no subject to scope to and no tenant to infer.
Reads see every tenant's rows; a write to a `tenant()` table without an explicit
`tenantId` fails with `TenantScopeViolation` rather than landing in an arbitrary
tenant. When your reaction is per-tenant, take the tenant from the row that
changed, as above. This is the same posture a schedule's `ctx.app.store` has —
both are post-request system work.

Writes are mixin-stamped exactly as a request-path write is: id scheme,
timestamps, audit columns. Only the subject differs.

## What fires when

The framework binds to the store's `onChange` channel. Subscribers fire **after the transaction commits** — the row IS persisted when your handler runs. This means:

- **You can read the just-committed row** via `event.new` (it's the committed value).
- **You CAN'T reject the change** — by the time the handler runs, the transaction has landed. For pre-write rejection, use [`table().validate(Schema)`](/docs/database/columns#table-level-validation).
- **No transaction boundary.** Side effects you do in the handler are not rolled back if some later operation fails.

The `on` filter narrows by operation:

- `'any'` (default) — every insert/update/delete fires the handler
- `'insert'` / `'update'` / `'delete'` — single op
- `['insert', 'delete']` — array of ops

Other-table events get filtered out before your handler sees them. The matcher does this at the dispatcher level so subscribers add zero hot-path overhead to writes that don't match their table.

## Semantics — best-effort, fire-and-forget

Subscribers are **non-durable** by design:

- **A throw doesn't fail the request.** The original mutation has already committed. The framework logs the failure (scoped to `subscribe:<filename>`) and the next event still arrives.
- **Async handlers are NOT awaited by the dispatcher.** Fire-and-forget — a slow handler can't back-pressure the change stream. Errors propagate to the structured log via `.catch()`, but the change-emission path returns immediately.
- **No retry, no resume.** If the process crashes mid-handler, the work is gone. Same if the network call inside the handler fails — there's no built-in retry policy.

If you need any of those properties (transactional, durable, retried), don't hand-roll the kickoff in the handler — a subscriber's `ctx` carries no workflow handle, deliberately. "A row changed → start a workflow" has its own primitive: a [reaction](/docs/data/reactions). A `*.reaction.tsx` watches the same post-commit change stream and its `act` starts the workflow for you, behind mandatory guards:

```tsx
// reactions/fulfillOrder.reaction.tsx
import { defineReaction } from '@voltro/runtime'

export default defineReaction({
  name:  'fulfillOrder',
  watch: { table: 'orders', on: 'insert' },
  act:   { kind: 'workflow', workflow: 'orders.fulfill' },   // the changed row is the payload
  guards: {
    // REQUIRED — a stable idempotency key, so the same change acts exactly once.
    dedupeKey: (event) => String((event.new as { id?: string })?.id ?? ''),
  },
})
```

The workflow body owns durability; the reaction owns the trigger and the guards. `dedupeKey` is not optional — an ungated reaction whose act writes back into the watched table self-triggers forever, so `defineReaction` throws at boot when it's missing. See [Reactions](/docs/data/reactions) for the `when` predicate, `rateLimit`, `costBudgetUsd`, and the `act: { kind: 'agent' }` target.

## Why this and not a regular subscription?

A `useSubscription('app', 'todos.list')` is the right tool when the **client** wants live data — the framework pushes a delta over WebSocket and React re-renders. Subscribers are the server-side equivalent: when something **on the server** wants to react to a write — emit a notification, refresh an index, push to a queue — without round-tripping through a connected client.

| What's reacting | Use |
|---|---|
| The browser, to show fresh data | `useSubscription` in a hook |
| A server-side action, every commit | `*.subscribe.ts` (this) |
| A server-side action, eventually-consistent + durable | [`*.reaction.tsx`](/docs/data/reactions) with `act: { kind: 'workflow' }` |
| A server-side action, in the same transaction | Do the work inside the mutation handler |

## Decision tree — picking the right seam

| What you want | Use |
|---|---|
| In-transaction work tied to a write | mutation handler |
| Reject the insert | [`table().validate(Schema)`](/docs/database/columns#table-level-validation) |
| Derive a value at write time | [`column.computed(row => ...)`](/docs/database/columns#computed-columns) / [`column.default(() => ...)`](/docs/database/columns#default-with-callback) |
| **Best-effort post-commit reactivity per row** | **`*.subscribe.ts`** (this) |
| Crash-safe post-commit reactivity | [`*.reaction.tsx`](/docs/data/reactions) acting on a workflow |
| Periodically-refreshed pre-computed query | [`*.aggregate.ts`](/docs/data/aggregates) |
| Event ingestion + analytical aggregates | [Analytics sink](/docs/plugins/analytics) |

## Common patterns

### Refresh a search index

```ts
import { meiliClient } from '../lib/meili'

export default defineSubscriber({
  table: 'posts',
  on:    ['insert', 'update'],
  handler: async (event) => {
    if (!event.new) return
    await meiliClient.index('posts').updateDocuments([event.new])
  },
})
```

### Mirror to an external system

```ts
export default defineSubscriber({
  table: 'organizations',
  on:    'insert',
  handler: async (event, ctx) => {
    if (!event.new) return
    const slug = event.new.slug as string
    // Best-effort sync; failure logs + continues
    await fetch('https://my-crm.example/sync', {
      method: 'POST',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ slug, name: event.new.name }),
    })
  },
})
```

### Soft-delete cleanup workflow

Starting a workflow off a state transition is a [reaction](/docs/data/reactions), not a subscriber — `when` expresses the transition, `dedupeKey` keeps it from acting twice:

```tsx
// reactions/cleanupDeletedDocuments.reaction.tsx
import { defineReaction } from '@voltro/runtime'

export default defineReaction({
  name:  'cleanupDeletedDocuments',
  watch: { table: 'documents', on: 'update' },
  // Fire ONLY when deletedAt was just set — not on every other update.
  when:  (event) => event.old?.['deletedAt'] === null && event.new?.['deletedAt'] != null,
  act:   { kind: 'workflow', workflow: 'documents.cleanupDeleted' },
  guards: { dedupeKey: (event) => String((event.new as { id?: string })?.id ?? '') },
})
```

The workflow does the heavy lift (scrub child rows, notify owners, archive to cold storage); the reaction's job is just to detect the state transition and hand the row over.

## Failure semantics in detail

The framework logs subscriber failures via the structured logger — scope `subscribe:<filename>` — so you can grep + filter with `voltro logs --scope 'subscribe:*'`. Operations the handler reaches that themselves emit logs (HTTP client, store calls) keep their own scope.

There's no retry. If a transient failure should be retried, do one of:

1. Wrap the call in your own retry policy inside the handler (`Effect.retry`, `pRetry`, etc.).
2. Move the work into a [reaction](/docs/data/reactions) whose `act` starts a workflow — workflows have first-class retry semantics, and the reaction's `dedupeKey` makes the trigger idempotent.

## By design

The subscriber is a deliberately thin post-commit hook — the sharp edges below are choices, not gaps. When you outgrow them, the escape hatch is a [reaction](/docs/data/reactions) acting on a workflow (durable, retried) or handler-side logic.

- **One subscriber per file.** Multiple defaults exported don't compose; pick the most natural file boundary (one cohesive concern per file).
- **Filter on `on` + table; per-row predicates live in the handler.** A predicate like "fire only when the user's plan changed" is a one-line guard at the top of the handler — kept there rather than in a framework pre-filter so the matching rule sits next to the code that reacts to it.
- **No batching.** Each commit fires its subscribers individually. If a hot write path needs batched network calls, batch inside the handler (rolling window, debounce) — the framework doesn't impose a batching window you'd have to fight.

## Configuration

`*.subscribe.ts` files are discovered + bound automatically by `voltro dev` / `voltro start`. There's no `app.config.ts` flag — the file existing IS the registration.

Subscribers fire AFTER the mutation handler commits, which happens AFTER any plugin `interceptMutation` chain returns. They're at the very tail of the write path:

```
client → AuthMiddleware → plugin interceptors → mutation handler → commit → subscriber handler
```

The subscriber sees the post-commit row, which has been auto-stamped by audit/tenant/softDelete mixins, validated against `table().validate(...)` if set, and persisted to the database.

## See also

- [Subscriptions](/docs/data/subscriptions) — the CLIENT-side `useSubscription` hook for live query data (different concept, similar name).
- [Streams](/docs/data/streams) — transient element feeds; for crash-safe post-commit work, use a [reaction](/docs/data/reactions) that acts on a workflow.



---

<!-- source: en/data/reactions.md -->
## Reactions

_Standing reactive agents — a *.reaction.tsx watches a table and runs an agent/workflow on a change behind mandatory spend guards (dedupe / rate-limit / budget)._

# Reactions (`*.reaction.tsx`)

A **reaction** watches a table and, on a relevant change, runs an agent or
workflow — behind MANDATORY spend guards. It's the data-driven sibling of a
[subscriber](/docs/data/subscribers): where a subscriber is "run my code on a
change," a reaction is "run an agent/workflow on a change, without footgunning a
spend storm." Discovery + dispatch are automatic — drop a `*.reaction.tsx` file
in and `voltro dev` binds it to the post-commit change stream.

```tsx
// reactions/flagBigOrder.reaction.tsx
import { defineReaction } from '@voltro/runtime'

export default defineReaction({
  name:  'flagBigOrder',
  watch: { table: 'orders', on: 'insert' },                 // table (+ optional op filter)
  when:  (e) => Number((e.new as { total?: number })?.total ?? 0) > 1000,  // optional predicate
  act:   { kind: 'workflow', workflow: 'orders.review' },   // run THIS workflow on the change
  guards: {
    // REQUIRED — a stable idempotency key so the same change acts ONCE. Without
    // it, a reaction whose act writes the watched table self-triggers into a
    // spend storm; defineReaction throws at boot if it's missing.
    dedupeKey:     (e) => String((e.new as { id?: string })?.id ?? ''),
    rateLimit:     { limit: 10, windowMs: 60_000 },         // optional per-reaction cap
    costBudgetUsd: 5,                                        // optional per-tenant AI ceiling
  },
})
```

## How it fires

Both `voltro dev` and `voltro serve` discover every `*.reaction.tsx`, bind it to
the store's post-commit `onChange` channel, and on each change run the act
through the guard chain: **op/table filter → `when` predicate → dedupe →
rate-limit → budget**. The changed ROW is handed to the act — as the workflow's
payload (so its payload schema should match the watched table's row), or seeded
into the agent's prompt.

## Guards (the point)

- **`dedupeKey` (required)** — the same logical change acts exactly once. This is
  what stops a reaction whose act writes the watched table from self-triggering
  forever. `defineReaction` throws at boot if it's missing.
- **`rateLimit` (optional)** — at most `limit` firings per `windowMs`.
- **`costBudgetUsd` (optional)** — a per-tenant AI spend ceiling; over budget,
  the reaction refuses (fails closed).

## The two act targets

- **`act: { kind: 'workflow', workflow }`** — starts the workflow on the change
  (durable, retryable; the row is the payload). Reach for this when you need
  durable multi-step orchestration.
- **`act: { kind: 'agent', agent }`** — fires the agent headlessly: opens a fresh
  thread, seeds it with the change as the prompt, and drives the synthesized
  `<agent>.send` as the agent actor (tenant-scoped to the row). The assistant's
  turn streams onto the durable `<agent>.messages` thread. Reach for this when a
  standing agent should react in natural language.

## Limits (v1)

- Best-effort + fire-and-forget (like subscribers) — a failing act logs +
  continues; it can't back-pressure the change stream. Durability comes from a
  workflow act (an agent act is best-effort).
- `dedupeKey` is in-memory per process in v1 (it stops the self-trigger storm
  within a run); a durable cross-restart dedupe table is a follow-up.

## When to use what

- **Run my own code on a change** → [`*.subscribe.ts`](/docs/data/subscribers).
- **Run an agent/workflow on a change, with spend guards** → `*.reaction.tsx` (this).
- **Reject a write / derive a value at write time** → schema DSL
  (`validate(Schema)` / `computed` / `default`).



---

<!-- source: en/data/approvals.md -->
## Approvals

_requiresApproval — a mutation or action that needs a second human before it takes effect, with the pending intent in a durable row and self-approval refused._

A mutation can declare that **one person is not enough**:

```ts
// apps/api/mutations/invoices.refund.mutation.ts
import { defineMutation } from '@voltro/protocol'
import { Schema } from 'effect'

export const refundInvoice = defineMutation({
  name: 'invoices.refund',
  input:  Schema.Struct({ invoiceId: Schema.String, amountCents: Schema.Number }),
  output: Schema.Struct({ ok: Schema.Boolean }),
  // The requester still has to be allowed to ASK.
  guards: [{ scope: 'invoices:refund' }],
  requiresApproval: {
    approvers: [{ scope: 'invoices:approve' }],
    expiresIn: '4h',
    reason: 'refunds move money out of the account',
  },
})
```

That is the whole declaration. The framework does the rest: the first call is
recorded and refused, a second human decides, and the identical call then
succeeds exactly once.

Human-in-the-loop already existed *inside* a durable workflow
([`awaitSignal`](/docs/workflows/definition), AI-Flows' human step). This is the
same idea for an **ordinary rpc call** — no workflow around it, no status column
to hand-roll, and the "who may approve" rule expressed in the same `guards:`
vocabulary as everything else.

## What the caller sees

The first call does **not** run. It fails with a typed `ApprovalRequired`:

```ts
import { errorTag } from '@voltro/protocol'

const result = await refund({ invoiceId, amountCents })
// throws:
// {
//   _tag: 'ApprovalRequired',
//   approvalId: 'apv_01j…',
//   procedure: 'invoices.refund',
//   expiresAt: '2026-08-12T14:00:00.000Z',
//   requiredScopes: ['invoices:approve'],
//   reason: 'refunds move money out of the account',
//   created: true,   // false when an earlier identical call already asked
// }
```

It is a typed failure rather than a success with a status field on purpose: a
mutation that returned its normal output shape when nothing happened is the
easiest thing in the world for a client to mis-handle, and every client already
branches on `_tag`.

For a mutation the **transaction never opens**. For an action the executor's
external I/O never happens — which is the only point at which nothing has
happened yet, since there is no rollback for an outbound HTTP call.

## The two built-in procedures

`__voltro.approvals.pending` is a reactive query over the approvals table, so
both sides of the exchange are live with no polling:

```tsx
import { useSubscription, useMutation } from '@voltro/client'
import type { PendingApproval } from '@voltro/protocol'

export const ApprovalQueue = () => {
  const { data } = useSubscription<ReadonlyArray<PendingApproval>>(
    'app', '__voltro.approvals.pending', {},
  )
  const decide = useMutation('app', '__voltro.approvals.decide')

  return (
    <ul>
      {(data ?? []).map((a) => (
        <li key={a.id}>
          {a.procedure} — {a.relation === 'to-decide' ? 'awaiting you' : `you asked · ${a.status}`}
          {a.relation === 'to-decide' && (
            <button onClick={() => decide.mutate({ approvalId: a.id, decision: 'approve' })}>
              Approve
            </button>
          )}
        </li>
      ))}
    </ul>
  )
}
```

Each row carries a `relation`: `'to-decide'` (you may act on it) or
`'requested'` (you asked for it). The requester watches their own row flip
`pending → approved` and re-fires the mutation; the approver's queue appears
without a refresh.

The feed is **scoped in the handler, not by a descriptor guard** — a row appears
only if you requested it or satisfy its recorded approver scopes, so an
anonymous caller sees an empty list. There is no scope that means "may see my own
approval work", and inventing one would be a guard that reads as protection and
enforces nothing.

## Where the pending intent lives, and why its identity matters

Between the request and the decision the intent is a row in `_voltro_approvals`
— durable, so it survives a restart, a rolling deploy and a replica switch.

Its identity is **content-addressed**: a digest over the procedure, the
requester, the tenant, the canonicalised input, and an optional caller `nonce`.
Both directions of getting that wrong are real bugs:

- too **coarse** (keying on the procedure, say) and two different pending
  intents share one row, so approving one executes the other's payload;
- too **fine** (a fresh id per attempt) and every page refresh, client re-send or
  transaction replay mints a second approval, asking the human twice for one
  decision.

Content-addressing is the only spelling that is stable across a retry and
distinct across intents. A `UNIQUE` on that key enforces **at most one live
intent per content**. Two deliberately identical requests — the same user really
does want to refund the same invoice twice — are expressible by passing a
different `nonce`, which is a decision you state rather than one the framework
guesses.

An approval is **consumed** when it admits a call. A replay after that is a new
request, not a free second execution.

## The refusals

These are the point of the feature, so they are worth reading as a list.

**Self-approval is refused, unconditionally.** There is no opt-out flag. The
whole content of "a second human" is that it is a second one, and a framework
that shipped `allowSelfApproval: true` would be shipping a control every team
turns off under deadline pressure — with the audit row still reading "approved".

The check runs **before** the authority check, deliberately: a requester who
happens to hold the approver scope is told they cannot approve their own request,
which is the accurate reason, instead of being quietly let through.

**An unauthorised approver** gets `ApprovalForbidden`, naming the scope they
lack. The check uses the same guard evaluator the dispatch spine runs, against
the scopes recorded **on the row** — including a resource-scoped guard's resolved
resource id, so `approvers: [{ scope: 'invoices:approve', resource: (i) => i.invoiceId }]`
stays scoped to that invoice at decision time rather than widening into a global
scope check.

**An anonymous decider** is refused. Every anonymous caller compares equal to
every other, so the identity the control rests on does not exist. For the same
reason, `requiresApproval` combined with `openAccess:` is refused at declaration.

**Expiry fails closed.** Past `expiresAt` the intent can be neither approved nor
executed — including an intent that was approved and then aged out before the
requester came back. The requester re-submits and a fresh decision is asked for.

**One intent, one verdict.** A second decision on the same intent gets
`ApprovalNotPending`.

A rejection is reported to the requester once, as `ApprovalRejected`, on their
next attempt. A further attempt after that opens a genuinely new decision — a
rejection is a verdict on one request, not a permanent ban on the operation.

## Declaration-time refusals

Two shapes read like a control and enforce nothing, so they throw where you can
still see both fields:

```ts
// ✗ nobody can approve this — every call would park forever
requiresApproval: { approvers: [] }

// ✗ an unauthenticated requester has no identity, so the self-approval
//   refusal cannot compare anything and the control degrades to nothing
openAccess: 'public', requiresApproval: { approvers: [{ scope: 'x' }] }
```

## Expiry as a tunable

Precedence: the descriptor's own `expiresIn`, then the app default, then the
environment, then 24 hours.

```ts
// app.config.ts
export default {
  approvals: { expiresIn: '4h' },
}
```

`VOLTRO_APPROVAL_EXPIRY_HOURS` overrides the built-in default. There is
deliberately no "never expires": an approval queue with no floor is a list of
decisions nobody made.

## Agent tools

`exposeAsTool: { confirm: true }` used to be a report — the inventory showed it
and nothing enforced it. It is now backed by this primitive: a confirm tool is
mountable only if its descriptor also declares `requiresApproval`, and then the
agent's call parks in your approval queue and returns `ApprovalRequired` to the
model. See [Agent tools](/docs/ai/tools).

## The table

`_voltro_approvals` is created for every app (it is one small table, and the
alternative would be a surprise `CREATE TABLE` on the production boot after
somebody adds `requiresApproval:` to a mutation). It is bounded by the retention
sweep — 30 days by default, `VOLTRO_APPROVALS_TTL_HOURS` to change it.



---

<!-- source: en/data/client-state.md -->
## Client state (defineStore)

_defineStore — client state that is not server state. Selector-only reads, key scoping instead of providers, and SSR seeding that rides the payload the router already writes._

Server state already has a home: a subscription **is** live server state, and it stays live. What has no home is the rest — which rows are selected, which wizard step you are on, the draft you have not submitted.

Without a primitive for that you reach for zustand or jotai, which is a *parallel runtime* — the one thing the framework asks you not to bring. So it ships one.

```ts
// wizard.store.ts
import { defineStore } from '@voltro/client'

export const wizard = defineStore('wizard', () => ({ step: 0, draft: '' }))
```

```tsx
const step = wizard.use((s) => s.step)                    // the global instance
const step = wizard.use((s) => s.step, { key: orderId })  // one instance per order

wizard.set({ step: 2 })
wizard.set((s) => ({ ...s, step: s.step + 1 }))
```

## Reads go through a selector — there is no `useStore()`

A component that holds the whole state re-renders on every change to any field, so an API that hands it over would be used and would be wrong. Reading a slice re-renders only when **that slice** changes:

```tsx
const coupon = wizard.use((s) => s.coupon)   // set({ note }) does not re-render this
const count  = cart.use((s) => s.items.length) // ['a'] → ['b'] does not re-render this
```

### Computed values need `equals: shallow`

A selector that *builds* something — an object literal, a mapped or filtered array — returns a fresh reference every call, so the default identity check reports "changed" forever and the component re-renders on **every** store change:

```tsx
import { shallow } from '@voltro/client'

const visible = cart.use((s) => s.items.filter((i) => i.visible), { equals: shallow })
const pair    = cart.use((s) => ({ a: s.a, b: s.b }), { equals: shallow })
```

You do not have to remember: in dev the framework detects the case and warns once, naming the fix. The selector is also memoised — it does not re-run while the state object is unchanged, so an expensive filter costs nothing on unrelated updates.

## Scoping is by key, not by a Provider

A Provider re-renders every consumer when its value identity changes, whether or not that consumer read the field that moved — that *is* what makes context painful at scale. So an instance is addressed by a key you already have:

```tsx
wizard.use((s) => s.step, { key: orderId })
```

which is the same model as everywhere else in the framework: `useSubscription('orders.list', { orgId })` is keyed by input, not by position in the tree. One read form, an optional key, no provider to forget.

`wizard.release(orderId)` drops an instance; `wizard.keys()` lists the live ones.

## SSR seeding adds no new channel

Call `seedStore` anywhere on the server during a render and the value reaches the client's first render:

```ts
export const loader = async ({ params }) => {
  seedStore(wizard, { step: 2 }, { key: params.orderId })
  return { /* … */ }
}
```

There is no `dehydrate()` to remember and no `hydrate()` to forget — the seed rides the hydration payload the router already writes, and `mount()` applies it before the tree exists. A step you can forget is a step somebody will.

On the client `seedStore` **throws**. A silent no-op would leave the store empty in the browser and full on the server, and that surfaces as a hydration mismatch that reads like a React bug.

**The key is part of the address.** Seed with `{ key: orderId }` and read with `{ key: orderId }`. A component reading the global instance while a loader seeded a key gets the initial value — correct, and easy to trip over once.

## Surviving a reload

`persist` writes the state to `localStorage` (or `sessionStorage`) on every change and reads it back when the store is defined:

```tsx
export const filters = defineStore(
  'inbox:filters',
  () => ({ status: 'open', sort: 'newest', draft: '' }),
  {
    persist: {
      key: 'inbox:filters',
      storage: 'local',              // 'session' lasts the tab
      pick: (s) => ({ status: s.status, sort: s.sort }),
      migrate: (stored) => (isFilters(stored) ? stored : undefined),
    },
  },
)
```

Three details are the whole reason this is in the framework rather than in your codebase, because a hand-rolled version gets all three wrong:

**The stored value is merged over `initial()`, not substituted for it.** Ship a new field and every returning user has state without it — `undefined` where the type promises a string. Merging means an old payload gains the new defaults.

**`migrate` returning `undefined` discards the value.** A stale draft is an annoyance; a half-migrated one is a bug report nobody can reproduce. Discarding is the right answer far more often than guessing, so it is the easy one to write.

**Every storage touch is guarded and wrapped.** The module is imported by the server render too, and Safari in private mode throws on *reading* `localStorage`, not just on writing. A store that throws at import time takes the page with it.

**A persisted store on a server-rendered page hydrates against the *server* value.** The server has no `localStorage`, so it renders `initial()`; the stored value lands in the commit right after hydration. Without that, every returning user would get a hydration mismatch — a flash and a console error that reads like a React bug. `get()` is not deferred, only the render: an action reading the draft before the first paint reads the draft.

`pick` narrows what gets written — persist the filters, not the open/closed state of every panel. And only the **global** instance persists: a keyed instance is per entity, and writing every key into one bucket grows without bound. Persist a map yourself if you mean to.

## Actions that write more than once

An action rarely touches one field. Applying a coupon writes the coupon *and* the recomputed total; that is one thing the user did, and `batch` says so:

```ts
checkout.batch('applyCoupon', () => {
  checkout.set({ coupon })
  checkout.set({ total: recompute(coupon) })
})
```

One notification, one entry in the devtools feed named `applyCoupon`, and **one undo step**. Without it the same action is three of each — Ctrl-Z walks back through a third of a change at a time, and the feed shows three anonymous writes instead of the thing that happened. React batches the re-*renders* on its own; it cannot batch the meaning.

**If the callback throws, every write it made is rolled back.** Nothing was announced yet, so an action that fails halfway cannot leave the half-applied state that is the usual reason people reach for a transaction. A nested `batch` joins its parent rather than opening a second one.

### Async work goes around the batch, not inside it

```ts
const quote = await fetchQuote(coupon)          // await FIRST
checkout.batch('applyCoupon', () => {           // then batch the writes
  checkout.set({ coupon, total: quote.total })
})
```

Passing an `async` function to `batch` is an **error**, not a warning. Everything after the first `await` would run outside the batch — writes escaping one at a time, a rollback covering only the synchronous head, and a devtools entry that lies about what the action did. Holding a batch open across time is not available: it would have to block every other write for the duration.

## Undo and redo

Every write passes through one seam, so the previous state is already recorded — undo is a lookup rather than a feature the store had to be designed around:

```tsx
draft.undo(orderId)      // back one write
draft.redo(orderId)      // forward again
draft.canUndo(orderId)   // for disabling the button
draft.canRedo(orderId)
```

It is a cursor over an intact history, not a stack that consumes entries. So repeated calls walk back through the steps rather than toggling between the last two, and a **new write after an undo drops the redo tail** — the behaviour every editor has.

An undo never becomes undoable itself, and each keyed instance has its own history.

## Devtools: inspect, and travel

The `voltro dev` overlay has a **Stores** tab. It lists every defined store with its live state — global and per key — and a feed of every write: which store, which key, the label if you passed one, and the fields that actually changed.

`◀ Back` and `Forward ▶` step through that feed, restoring the state as it was before or after each write. The state a component reads moves with it, so you can walk back to the moment before a bug and watch it happen again.

No extension, no connector, no version to match. Every write already passes through one seam, so the panel is just another subscriber — it sees writes made by code that never heard of devtools, on any machine, including a colleague's.

## What a store must never hold

Server data. Copying a subscription's rows into a store gives you a second copy that does not live; the page then renders the stale one, and the bug presents as *"reactivity is broken"*. `voltro doctor` reports a `*.store.ts` that reads a subscription.

Read server state where you render it, and keep the store for what is genuinely client-side.



---

<!-- source: en/data/outbox.md -->
## Transactional outbox

_ctx.outbox.enqueue — a reliable external side effect from a mutation, committed in the same transaction as the write that caused it._

A mutation must not do external I/O. It runs in a transaction, and an HTTP call
cannot be rolled back — if the request succeeds and the transaction then fails,
you have charged a card for an order that does not exist.

So "write this row **and** sync it to Jira" has no correct one-step form. The
usual workaround is to build one: a deliveries table written inside the
transaction, a cron that drains it, and a worker with backoff and a dead-letter.
That is a real subsystem, and every integration app rebuilds it.

`ctx.outbox` is that subsystem, as a one-liner.

```ts
// apps/api/mutations/ticket.create.server.ts
export default async (input: { title: string }, ctx: AppContext) => {
  const ticket = await ctx.store.insert('tickets', { title: input.title })

  await ctx.outbox.enqueue('jira.sync', { ticketId: ticket.id }, {
    idempotencyKey: `jira.sync:${ticket.id}`,
  })

  return { id: ticket.id }
}
```

## Why this is correct, not just convenient

`enqueue` writes through `ctx.store` — and inside a mutation, `ctx.store` **is
the transactional view**. The outbox row commits in the same transaction as the
domain write, or neither does.

That is the whole guarantee. There is no window in which the ticket exists and
the intent to sync it was lost, because losing the intent means the ticket was
rolled back too.

This is what separates it from reacting to a change *after* commit. A
post-commit tap — including `@voltro/plugin-cdc-out`, which says so plainly —
is at-least-once **from enqueue**: a crash between the commit and the tap loses
the event. Here, enqueue cannot be lost.

Delivery *after* commit is still at-least-once. That is the strongest guarantee
available without a distributed transaction into the target system, so:

**Handlers must be idempotent.** A process that dies between "the remote
accepted it" and "we recorded that" will retry.

## Declaring the handler

One `*.outbox.ts` file per effect:

```ts
// apps/api/outbox/jira.sync.outbox.ts
import { defineOutboxHandler } from '@voltro/runtime'

export default defineOutboxHandler({
  effect: 'jira.sync',
  maxAttempts: 5,
  handler: async ({ payload, attempt, subjectId, traceId }) => {
    await jira.syncIssue(payload['ticketId'] as string)
  },
})
```

The handler runs **after** the enqueuing transaction committed, outside it, and
may do external I/O — that is the point. It receives the payload, the attempt
number, and the subject / tenant / trace of whoever enqueued it.

Two handlers claiming the same `effect` is refused at boot with both filenames,
rather than letting whichever loaded last silently win.

## Retries, backoff, dead-letter

| | |
| --- | --- |
| Retry schedule | exponential — 1s, 2s, 4s … capped at 5 minutes |
| Default attempts | 8 (`maxAttempts` on the handler, or per-enqueue) |
| Exhausted | row moves to `dead`, logged at ERROR, stays in the table |
| Unknown effect | left **pending**, never discarded |

That last row matters. The usual cause of an unknown effect is a deploy where
the enqueuing code shipped ahead of its handler. Dead-lettering those would turn
a rollout ordering detail into permanent loss of a side effect the app believes
happened, so they wait instead.

A dead-lettered row is not deleted — it is queryable in `_voltro_outbox` with
its `lastError`, because a dead letter is a side effect your app thinks occurred
and which never will.

## Delivery, and why there is both a nudge and a poll

When the transaction commits, the worker is nudged and the effect usually goes
out in milliseconds. A poll also runs every 5 seconds.

The nudge is an optimisation. The **poll is the contract**: it picks up rows
whose nudge was lost because the process died between commit and delivery, rows
enqueued by another replica, and rows waiting out a backoff. Without it the
guarantee degrades to "delivered unless something went wrong" — which is the
exact case a durable outbox exists for.

**The 5 seconds is a floor, not a rate.** Those three reasons are taken one at a
time, and only the last needs a clock:

| the poll exists for… | what brings the loop back |
| --- | --- |
| a nudge lost to a dead process | the first pass at boot, which is not deferred behind a tick |
| a row another replica enqueued | a change event on `_voltro_outbox` |
| a row waiting out a backoff | the runner arms for that row's own `nextAttemptAt` |

So on an empty queue the timer **stops entirely** and the loop waits to be woken.
Where no change channel is available it keeps the fixed 5-second tick instead —
the poll is then the only thing that can notice another replica's row, and a
durable outbox that stops looking is worse than one that polls.

## Options

```ts
await ctx.outbox.enqueue('mail.welcome', { userId }, {
  idempotencyKey: `welcome:${userId}`,  // drop if an undelivered row has this key
  maxAttempts: 3,                        // override the handler's default
  delayMs: 60_000,                       // don't attempt before then
})
```

`idempotencyKey` dedupes against rows that have not yet succeeded, so a retried
mutation does not produce a second side effect. A **delivered** key is
deliberately not a blocker — reusing a key later means "do it again", and
treating it as permanently consumed would silently swallow a legitimate request.

`enqueue` returns the outbox row id, which is also the delivery id: persist it
alongside your row and a client can watch the effect's progress.

## Delivery history

The queue answers "is this still owed". It cannot answer "what did the remote
say on attempt 3", "how long did it take", or "who resent it" — which is what a
delivery-history screen renders. So every attempt appends a row to
`_voltro_outbox_attempts`:

| column | |
| --- | --- |
| `outboxId` | the entry this attempt belongs to |
| `effect` | denormalised — the history stays readable after the entry is purged |
| `attempt` | 1-indexed, monotonic across the entry's whole life |
| `trigger` | `automatic` (the drain) or `manual` (a resend) |
| `triggeredBy` / `reason` | who asked for a manual resend, and why |
| `outcome` | `delivered` · `failed` · `dead` (`dead` = the attempt that exhausted the budget) |
| `startedAt` / `finishedAt` / `durationMs` | timing |
| `error` | failure message (clipped) |
| `response` | whatever the handler **returned**, as JSON |
| `subjectId` / `tenantId` / `traceId` | carried from the entry |

`response` is how per-attempt transport detail gets recorded without the
framework pretending to model HTTP — return `{ status, body }` from the handler
and the history has it:

```ts
export default defineOutboxHandler({
  effect: 'jira.sync',
  handler: async ({ payload }) => {
    const res = await fetch(url, { method: 'POST', body: JSON.stringify(payload) })
    if (!res.ok) throw new Error(`jira ${res.status}`)
    return { status: res.status, body: (await res.text()).slice(0, 500) }
  },
})
```

It is a framework table like any other, so reading it is a normal store read —
there is no separate history API to learn:

```ts
const history = await ctx.store.query({
  table: '_voltro_outbox_attempts',
  predicate: eq('outboxId', deliveryId),
  order: [{ column: 'attempt', direction: 'desc' }],
  take: 20,
})
```

Attempts are recorded best-effort: if the log write fails, the delivery still
succeeds. History must never be able to break the thing it observes.

## Resending on demand

```ts
await ctx.outbox.resend(deliveryId, { reason: 'customer never received it' })
```

Re-arms one entry for delivery now — including one that already reached `dead`.
This is legitimate precisely because delivery is already at-least-once and
handlers are therefore already required to be idempotent: a resend is the same
hazard the contract obliges them to absorb, not a new one.

What it must never be is invisible. The resulting attempt is recorded with
`trigger: 'manual'` plus the requesting subject and reason, so an operator
redelivery can never be mistaken for a backoff retry when someone later asks why
the remote saw the effect twice. The entry also carries a `resendCount`.

Three behaviours worth knowing:

- It **re-arms the existing entry** rather than enqueuing a copy. A copy would
  carry the same `idempotencyKey`, duplicate the payload, and split one entry's
  history across two ids.
- It grants a **small fresh budget** (`attempts`, default 1). A dead entry has
  already spent its allowance, so without this it would re-die untried — and
  "try again now" is what the button means, not "restart the whole schedule".
- It **refuses an entry whose attempt is in flight**, where re-arming would race
  the running attempt into a double delivery.

## Retention

Two bounds, because an unbounded attempt log is a slow-motion outage:

- A **per-entry cap** (50 attempts) trims the oldest as new ones arrive. The
  automatic path can't reach it — `maxAttempts` defaults to 8 — so this bounds
  the one case that is genuinely unbounded: an entry resent by hand for years.
  It works on every dialect.
- A **time bound** on the boot sweep purges attempts older than 30 days
  (`VOLTRO_OUTBOX_ATTEMPTS_TTL_HOURS`), plus **delivered** queue rows
  (`VOLTRO_OUTBOX_TTL_HOURS`). `dead` and `pending` entries are never aged out —
  a dead letter is an unresolved incident and a pending one is still owed.

The time sweep runs on postgres; on other dialects the per-entry cap is the
bound.

## When NOT to use it

- **Work that must be observable step-by-step, or that suspends** → a
  [workflow](/docs/workflows/overview). The outbox delivers one effect; it is
  not a durable multi-step process.
- **Reacting to any change on a table, not to one mutation's intent** →
  [`defineSubscriber`](/docs/data/subscribers) or
  [`defineReaction`](/docs/data/reactions).
- **Mirroring a table outward continuously** → `@voltro/plugin-cdc-out`, which
  is built for reverse-ETL with per-pipe ordering.

## There is no generic job queue — take X for Y

Voltro deliberately ships no `defineJob` primitive (priorities, worker pools, a
BullMQ equivalent). The outbox, [workflows](/docs/workflows/overview) and
[schedules](/docs/scheduling/overview) cover the cases between them, and a third
durability primitive would drift from both. What to reach for instead:

| you want… | take |
| --- | --- |
| a concurrency-limited worker pool | workflows + [declarative flow control](/docs/workflows/declarative-flow-control) — `concurrency` / `throttle` bound how many runs execute at once |
| true priority scheduling (high-priority work overtakes queued low-priority work) | does not exist as a primitive — a workflow draining **your own queue table** in your priority order is the honest build |
| a delayed / scheduled message | `delayMs` on `enqueue` (above) for a one-off delayed effect; a [schedule](/docs/scheduling/overview) for recurring time-based work; workflow [`sleep`](/docs/workflows/sleep) for a pause inside a durable process |
| exactly-once delivery | does not exist — delivery is at-least-once everywhere, which is the strongest guarantee available without a distributed transaction into the target; **idempotent handlers are mandatory** (see above) |

## See also

- [Mutations](/docs/data/mutations) — the transaction boundary this rides
- [Subscribers](/docs/data/subscribers) — post-commit reactions to table changes
- [Workflows](/docs/workflows/overview) — durable multi-step work



---

<!-- source: en/data/wire-protocol.md -->
## Wire protocol

_What's on the WebSocket — @effect/rpc over JSON, the snapshot/delta subscription envelope, and the POST /rpc one-shot path._

Voltro's client and api speak `@effect/rpc` over a WebSocket. Every primitive — queries, mutations, actions, agents, subscriptions — rides the one bidirectional connection, multiplexed by the rpc layer. The serialization is **JSON** (`RpcSerialization.layerJson`), not a bespoke binary format.

You usually don't think about the wire — the typed client + `@voltro/protocol` handle it end-to-end. This page is for when you DO need to: debugging a mystery, proxying through a gateway, or understanding what the inspect tooling shows you.

## The transport

The api boots two rpc server instances on the same `HttpLayerRouter`:

- **WebSocket** — `RpcServer.layerProtocolWebsocketRouter({ path })`, the primary transport. Long-lived; carries streaming queries (subscriptions), unary mutations/actions, and agent runs.
- **HTTP one-shot** — `RpcServer.layerProtocolHttpRouter({ path: '/rpc' })`, registered at `POST /rpc`. Non-streaming; one request → one batched response.

Both are provided `RpcSerialization.layerJson` and the SAME rpc group (`options.group`), so the SAME per-rpc handlers + the group-level `AuthMiddleware` / `ConnectionInfoMiddleware` run on either path. A forwarded session cookie resolves the same `Subject` + tenant whether the call arrives over WS or HTTP.

Because `@effect/rpc` owns the framing, there's no app-level frame taxonomy to learn — the rpc client encodes a request, the server decodes it, runs the handler, and streams back the result. What's app-specific is the **payload schema** of each rpc (your `defineQuery` / `defineMutation` input + output) and, for streaming queries, the subscription-event envelope below.

## Subscription events — snapshot / delta

A streaming query (what `useSubscription` opens) emits a sequence of **subscription events**. The envelope is defined in `@voltro/protocol`'s `subscriptionEvent(output)` — a `Schema.Union` of three variants, parameterised on the query's declared `output` schema so the rpc layer enforces the row shape end-to-end:

```ts
// snapshot — always the FIRST event; the full initial query result
{ _tag: 'snapshot', revision: number, data: <output> }

// delta — every subsequent event; an id-keyed JSON-patch, NOT full data
{
  _tag: 'delta'
  revision: number
  emittedAt: number
  patch: {
    ops: Array<
      | { op: 'add';     path: '/<id>'; value: <row> }
      | { op: 'replace'; path: '/<id>'; value: <row> }
      | { op: 'remove';  path: '/<id>' }
    >
    order: Array<id>   // the full id sequence of the next set, in order
  }
}

// error — this ONE subscription's handler failed; surfaced in-band on its own
// stream (never a defect that would stall siblings on the shared connection)
{ _tag: 'error', error: { _tag?: string, message: string, ...fields }, revision?: number }
```

- **`revision`** — monotonically increasing; lets the client order events and detect gaps.
- **`emittedAt`** — epoch milliseconds, present on `delta` only.
- **`data`** (snapshot) — the full payload, typed by the query's `output` schema (a row, an array of rows, a computed value — whatever the handler returns).
- **`patch`** (delta) — an id-keyed RFC-6902-style patch against the row set the client last held.
- **`error`** — a JSON-safe shape of a typed error (its `_tag` + fields + `message` preserved, so the client can pattern-match `error._tag`). `useSubscription` exposes it as `.error` for THAT query key.

The first event is always a `snapshot` carrying the full initial result — the client materialises it as its base. Subsequent events are `delta`s carrying **only the rows that changed**, plus the next id `order`. The server still re-runs the query on a change (the patch saves wire egress, not the re-query); it then diffs the previous row set against the new one into the patch.

### Per-subscription errors — isolated, not connection-wide

Many subscriptions multiplex over ONE WebSocket. If a single subscription's
handler fails (e.g. a live single-row getter that throws for a stale/foreign
id), the server emits an **`error` event on THAT subscription's own stream** and
completes it — it does **not** let the failure become a defect, which would
propagate to the shared connection and stall every *other* subscription on it
(the classic "one not-found and the whole dashboard hangs on loading"). The
client surfaces it as `useSubscription(...).error` for that one query key;
siblings keep delivering their snapshots and deltas.

**Author a live-subscribed getter to return, not throw.** A subscription is a
long-lived stream, so a getter that throws on every re-evaluation is a broken
stream. For an expected-absent row, make the query `output: Schema.NullOr(...)`
and return `null` — that's a normal snapshot the widget renders as "empty",
cleaner than an error banner. Reserve throwing for genuinely exceptional cases;
even then it's now contained to the one subscription.

### How the patch is keyed — by row `id`, not array index

The diff keys on each row's `id`, addressing it as `path: '/<id>'`, rather than by array index. Index-based paths are brittle the moment a row moves — and reordering is the common case for the live result sets this targets (leaderboards, collaborative lists, game state). Id-keying makes a reshuffle cost just the changed rows plus the id list:

- **`replace`** — a row present in both prev and next whose content changed; `value` is the full new row.
- **`add`** — a row new in next; `value` is it.
- **`remove`** — a row gone from next.
- **reorder** — carried entirely by `order` (the next id sequence); a pure reorder emits **zero ops** and just a new `order`.

The client keeps the last materialised row set, applies the ops to an id→row map, then materialises the result strictly in `order`. The round-trip is exact: applying the computed patch to the previous set reproduces the new set for every case — add, remove, replace, reorder, and combinations.

**No-op suppression still applies.** When a change re-runs the query but the resolved row set is unchanged, no event is emitted at all — the patch path covers "a small part of a big result changed", the suppression covers "nothing changed".

**Fallbacks.** A result set whose rows aren't id-keyed (a custom projection that drops `id`) can't be diffed by id, so its update ships as a full `snapshot` instead. Computed queries (a handler that returns a scalar or aggregate, not an id-keyed row set) likewise ship every update as a `snapshot` — there's no row set to patch.

## HTTP one-shot rpc (`POST /rpc`)

The WebSocket is the primary transport, but the api ALSO exposes `POST /rpc` for non-streaming invokes. It speaks the same JSON envelope and runs through the SAME per-rpc handlers + the same auth middleware — a forwarded session cookie resolves the same Subject + tenant as the WS path.

This is what a server-side web-router loader's `ctx.query` uses: a loader runs inside the request handler with no WS connection, so it calls the backend over HTTP for SSR first-paint + `meta`. The HTTP protocol drains a streaming query handler's stream to completion and returns it, so a streaming query yields its FIRST (initial) `snapshot` in the response batch — exactly what first-paint needs. See [Loaders & meta](/docs/routing/loaders-and-meta#fetching-backend-data-with-ctx-query).

For live, after-hydration data, subscribe over the WebSocket with `useSubscription` instead — `POST /rpc` is one-shot and never streams deltas.

## Authentication

The session cookie travels in the WebSocket upgrade headers (and in the `POST /rpc` request headers). The api's `AuthMiddleware` resolver decodes it and binds the resolved `Subject` to the connection; subsequent rpc calls on that connection inherit it.

For API-key authentication, send `Authorization: Bearer <key>` in the upgrade (or request) headers — the same resolver path maps it to a `subject.type === 'apiKey'`.

## Debugging

Use the framework's own tooling rather than reading raw frames:

- **`voltro traces`** — the per-request span waterfall, including each subscription `snapshot` / `delta` delivery with its produce→push latency. `voltro traces --errors` filters to failed hops.
- **`voltro logs --trace <id>`** — every log line of one request across hops, in order.
- The inspect dashboard's stream firehose (`GET /_voltro/inspect/stream`) surfaces rpc / cdc / log events live.

## Anti-patterns

- **Hand-crafting rpc frames.** Use `@voltro/client` — `@effect/rpc` owns the framing and the protocol can change.
- **Long-polling fallbacks.** None — Voltro is WS-or-bust for live data. Browsers without WebSocket support don't get reactive updates (one-shot reads still work over `POST /rpc`).
- **Proxying through a CDN without WebSocket support.** Cloudflare / Fastly / Vercel Edge all support it; configure the upgrade headers.



---

<!-- source: en/data/errors.md -->
## Error handling

_Schema-tagged error variants, runtime errors vs business errors, client narrowing, retry semantics._

Voltro distinguishes **business errors** (typed, declared in the schema, expected) from **runtime errors** (unexpected exceptions, transport failures, framework bugs). Each surface has clear semantics.

Live — `demo.greet` throws a typed `NameTooLong` for a name over 20 chars; the
client narrows on `_tag` and reads the typed fields:

```tsx
const greet = useAction('app', 'demo.greet')
try { await greet.run({ name }) }
catch (e) { if (e._tag === 'NameTooLong') { /* typed: e.max, e.actual */ } }
```

## Business errors

Declare your own with `Schema.TaggedError` — the class-extension form:

```ts
import { Schema } from 'effect'

class InsufficientFunds extends Schema.TaggedError<InsufficientFunds>()('InsufficientFunds', {
  required:  Schema.Number,
  available: Schema.Number,
}) {}

class TitleTooLong extends Schema.TaggedError<TitleTooLong>()('TitleTooLong', {
  maxLength: Schema.Number,
}) {}
```

Wire them into a mutation / action / query via the `error` field:

```ts
export const transfer = defineMutation({
  name:   'wallet.transfer',
  guards: [{ scope: 'wallet:transfer' }],
  input:  Schema.Struct({ to: Schema.String, amount: Schema.Number }),
  output: Schema.Struct({ txId: Schema.String }),
  error:  Schema.Union(InsufficientFunds, TitleTooLong),
})

export default async (input, ctx) => {
  const wallet = await ctx.store.select('wallets').where('id', ctx.request.subject.id).one()
  if (wallet.balance < input.amount) {
    throw new InsufficientFunds({ required: input.amount, available: wallet.balance })
  }
  // …
}
```

Client narrows on `_tag`:

```tsx
const transfer = useMutation('app', 'wallet.transfer')

const result = await transfer.mutate({ to: 'usr-42', amount: 100 }).catch((e) => e)

if (result._tag === 'InsufficientFunds') {
  // result.required, result.available are typed
  toast.error(`Need ${result.required}, have ${result.available}`)
} else if (result._tag === 'TitleTooLong') {
  toast.error(`Too long — max ${result.maxLength}`)
} else if ('txId' in result) {
  toast.success(`Transfer ${result.txId} complete`)
}
```

### What a rejected write actually rejects with

**The value is the typed error itself** — the same value `useSubscription` reports on the read side. Not a wrapper around it, and not an Effect `FiberFailure` (whose `_tag` would be `undefined`, so every branch above would silently fall through to the generic one).

That guarantee covers every write hook, not just `mutate`:

| Hook | Rejecting call |
|---|---|
| `useMutation` | `mutate(input)` |
| `useAction` | `run(input)` |
| `useWorkflow` | `start` · `cancel` · `resume` · `signal` · `update` |
| `useWorkflowSignal` / `useWorkflowUpdate` | `signal` · `update` |
| `useUpload` | `upload` · `uploadMany` |

Two consequences worth knowing:

- A **defect** (an undeclared `throw` server-side, a transport drop) rejects too, but with a plain `Error` — so an `_tag` check on it is `undefined` and falls to your generic branch, which is the intended split. Check `_tag` for the outcomes you declared; treat everything else as unexpected.
- Passing `onError` (or `notify.error`) instead **resolves** with `undefined` and hands that same typed value to your handler. See [Mutations](/docs/data/mutations).

Tagged errors:

- Are wire-safe — they serialise as JSON (the rpc transport is `RpcSerialization.layerJson`) and restore on the client with the correct `_tag` + payload.
- Narrow correctly in TypeScript via the `_tag` discriminant.
- Carry typed payload fields.
- Don't fire alerts / unhandled-promise-rejection signals — they're expected business outcomes.

## Runtime errors

Unexpected throws — `TypeError`, `RangeError`, framework bugs, third-party SDK failures — don't match the descriptor's `error` union, so they surface as **defects** rather than typed failures:

- The throw is logged (via `@voltro/logger` to your configured sink).
- The client sees a generic failure (no typed payload — runtime errors might leak sensitive context).
- The trace records the throw + trace ID for cross-referencing; `voltro logs --trace <id>` returns the whole causal chain.
- The trace is searchable in OpenTelemetry.

To distinguish "we know about this" from "this surprised us":

| Throw | Treatment |
|---|---|
| `throw new InsufficientFunds({ … })` (declared tagged error in `error:`) | Marshalled to the client typed, no log-volume increase |
| `throw new Error('oops')` | Surfaces as a defect — full log + trace, generic failure to client |
| Unexpected exception from a library | Same — caught, logged, generic failure |

## What the framework merges into your `error:` union

You do not declare the errors the framework itself can raise for a procedure —
they are unioned into the wire contract for you, and only when the procedure can
actually produce them:

| Merged | Into | When |
|---|---|---|
| `ScopeError` | query · mutation · action · event | the procedure declares a real `guards:` entry. NOT for `openAccess:` — a procedure advertising a denial it cannot produce is what makes an error union stop meaning anything |
| `Unauthenticated` | query · mutation · action · event | the same condition. A guard can refuse for TWO reasons and they mean different things: the caller is known and lacks the scope (`ScopeError`), or they presented a credential that was REJECTED and so arrived anonymous (`Unauthenticated`). Without the second, an expired session reads as a permissions problem |
| `BusinessRuleViolation` | mutation | always. A cross-table `rule()` on any table the mutation writes can fail it, and the descriptor cannot know which tables carry rules |
| `ApprovalRequired` · `ApprovalExpired` · `ApprovalUnavailable` | mutation · action | the procedure declares `requiresApproval:` |

So a guarded mutation does **not** need `ScopeError` in its own `error:`. If you
declared it anyway, that is harmless — the union is the same either way.

## Framework-shipped error variants

`@voltro/protocol` exports the framework's own tagged errors. The rest of your typed errors are ones you declare yourself (above) or ones a plugin contributes.

**Import them from `@voltro/protocol`, not from `@voltro/runtime`** — even the store errors the runtime raises. A descriptor that declares one in `error:` is loaded value-level by the web client (the `RpcClient` needs every procedure's Schema), and `@voltro/runtime` is server-only, so a descriptor importing from it is refused at boot by the browser-safety guard. `@voltro/runtime` re-exports them for server code, which never sees the difference.

| Variant | From | When |
|---|---|---|
| `ScopeError` | `@voltro/protocol` | `requireScope(subject, scope)` failed — `{ required, message }`. |
| `Unauthenticated` | `@voltro/protocol` | The resolved Subject is anonymous but a signed-in caller was required — optional `{ reason }`. |
| `TenantScopeViolation` | `@voltro/protocol` | A tenant-scoped `EffectStore` write had no authenticated subject. |
| `TenantRowNotFound` | `@voltro/protocol` | A keyed-by-id write (`store.update(t, id, …)`, `delete`, `hardDelete`, `patchJson`) on a `tenant()` table found no such row **in the caller's tenant**. Raised identically whether the row is missing or belongs to another tenant — the distinction would be a cross-tenant existence oracle. |
| `ServerOnlyColumnWrite` | `@voltro/protocol` | A `crud.create` / `crud.update` input tried to set a [`.serverOnly()`](/docs/database/sensitivity) column — `{ table, columns }`. |
| `StoreOperationFailed` | `@voltro/protocol` | The underlying store operation failed (transient). |
| `TableValidationFailed` | `@voltro/protocol` | A `table().validate(Schema)` row check rejected the write. |
| `ConstraintViolation` | `@voltro/protocol` | The database refused the write on a foreign key / unique / NOT NULL / CHECK — `{ kind, table, operation, constraint?, column? }`. See below. |
| `CacheError` | `@voltro/cache` | A cache backend op failed — `{ operation, key, cause }`. |
| `RateLimited` | `@voltro/plugin-ratelimit` | The limiter rejected the call — `{ limit, retryAfterMs, resetAtMs }`. |
| `TenantMismatch` | `@voltro/plugin-multitenancy` | `assertOwnTenant(input.tenantId, subject)` rejected a cross-tenant write. |

Each plugin that ships an error (`RateLimited`, `EntitlementExceeded`, `StorageError`, `MailError`, …) merges it into every procedure's wire-error union, so the client decodes it typed without you adding it to each `error:`. To surface a store error to the client, add it to the descriptor's `error:` union yourself (e.g. `error: Schema.Union(TenantScopeViolation, StoreOperationFailed, MyDomainError)`).

All carry `_tag` + typed payloads, all narrow correctly on the client.

### When the database refuses the write

An integrity rule your schema declares — a `reference()`, a `.unique()`, a non-nullable column, a `.check()` — is enforced by the database, not by the handler. When it fires you get a `ConstraintViolation`:

```ts
{ _tag: 'ConstraintViolation',
  kind: 'foreignKey', table: 'tasks', operation: 'insert',
  constraint: 'tasks_laneId_fkey' }
```

`kind` is the field to branch on, and it is the one field every dialect can always fill:

| `kind` | Means | The caller's fix |
|---|---|---|
| `foreignKey` | the row you referenced does not exist | point at a real row |
| `foreignKeyInUse` | this row may not go — others still reference it | delete the children first, or don't delete |
| `unique` | a row with this value already exists | pick another value |
| `notNull` | the column requires a value | send one |
| `check` | the row does not satisfy a declared `.check()` | fix the value |

`constraint` and `column` are filled where the dialect names them. sqlite reports a foreign-key failure as the bare sentence `FOREIGN KEY constraint failed` — no name, no column — so `constraint` is absent there and the direction is inferred from the operation.

**It carries names, never the driver's sentence, and that is deliberate.** The message the database produces contains row DATA on most engines: postgres attaches `Failing row contains (…)` — the *complete* row, every column, [`.sensitive()`](/docs/database/sensitivity) ones included — to a not-null and a check violation; mysql and mssql echo the duplicate value on a unique violation. A constraint name is schema, the same class of fact `TableValidationFailed.table` already puts on the wire. A row is data, and the caller who provoked the error is not automatically entitled to it. The full driver text is in the server log, with the trace id.

Declare it in `error:` to pattern-match it:

```ts
error: Schema.Union(ConstraintViolation, MyDomainError),
```

If you do not, it still reaches the client — collapsed to `InternalError` like any undeclared error, but carrying its own sentence (`ConstraintViolation: foreign key tasks_laneId_fkey on tasks: the referenced row does not exist`) rather than the opaque `Failed to execute statement` a raw `SqlError` produces.

## Loader errors

Loaders run in the web app — they receive `{ params, query, headers, signal }`, NOT a server `ctx` with `.store`. A loader reaches the backend through `query(...)` (the `POST /rpc` path, server-side only). It short-circuits with the branded control-flow signals `NotFoundError` / `RedirectError`:

```ts
import { notFound } from '@voltro/web'

export const loader = async ({ params, query }) => {
  // `query` is present only server-side (SSR/ISR). Guard for client nav.
  const note = query ? await query('notes.get', { id: params.id }) : undefined
  if (!note) throw notFound(`note ${params.id}`)   // → scoped not-found.tsx / 404
  return note
}
```

A `NotFoundError` renders the scoped `not-found.tsx` subtree (it is NOT routed to the `error.tsx` boundary). A plain `throw new Error(...)` DOES hit `error.tsx`:

```tsx
export default function ErrorPage({ error, reset }: { error: Error; reset: () => void }) {
  return <GenericError error={error} reset={reset} />
}
```

See [Loaders & meta](/docs/routing/loaders-and-meta) for `NotFoundError` / `RedirectError` and the `ctx.query` SSR path.

## Retry semantics

| Surface | Auto-retry? | Notes |
|---|---|---|
| `useSubscription` | ✓ on disconnect | The api connection auto-reconnects with exponential backoff (`500ms × 2^(attempt-1)`, capped at 5s) and re-subscribes; the server replays the current snapshot. Reconnect attempts are not capped — it keeps trying until the connection is restored or the component unmounts. |
| `useMutation` | ✗ | Mutations might be non-idempotent. Retry only when the operation is safe to repeat. |
| `useAction` | ✗ | Same. |
| `useAgentStream` / `useAgent` | ✗ | Streaming is hard to resume. Use a workflow for durable agent runs. |
| Workflow steps | ✓ | Configurable per-workflow + per-step. |

For mutations that are safe to repeat, wrap `mutate` in your own retry:

```tsx
const result = await retry(
  () => create.mutate(input),
  { attempts: 3, backoff: 'expo' },
)
```

For operations that must be durable or exactly-once across disconnects, queue the work through a workflow and use database uniqueness constraints around the business key.

## Validation errors

Each rpc decodes its input against the declared `input` schema before the executor runs. A payload that fails the schema is rejected by `@effect/rpc` as a decode failure — the executor never runs. To surface field-level messages to a form UI, declare your own validation error and decode the input yourself in the handler:

```ts
import { Schema } from 'effect'

class ValidationFailed extends Schema.TaggedError<ValidationFailed>()('ValidationFailed', {
  errors: Schema.Array(Schema.Struct({ path: Schema.String, message: Schema.String })),
}) {}
```

You can attach custom messages to the field constraints with `Schema.message`:

```ts
Schema.String.pipe(
  Schema.minLength(1, { message: () => 'Title is required' }),
  Schema.maxLength(200, { message: () => 'Title is too long (max 200 chars)' }),
)
```

Schema-level row validation on a table (`table().validate(Schema)`) throws the runtime's `TableValidationFailed` — declare it in the mutation's `error:` to surface it typed.

## Network errors

A WebSocket disconnect interrupts in-flight unary calls — the awaited `mutate` / `run` rejects. The framework reconnects query subscriptions automatically; for mutations and actions, you decide whether retrying is safe:

```tsx
try {
  await create.mutate(input)
} catch (e) {
  // Connection dropped mid-call. Wait for the api to reconnect, then retry
  // ONLY if the operation is idempotent.
  await waitForReconnect()
  await create.mutate(input)
}
```

## Anti-patterns

- **Swallowing every error and showing "Something went wrong".** You're hiding genuine bugs. Let the dashboard's error pane + `voltro traces --errors` surface them; only catch the specific tagged variants you've declared.
- **Throwing strings.** `throw 'oh no'` → surfaces as a defect, no useful payload. Use `Error` subclasses or `Schema.TaggedError` variants.
- **`if (error instanceof InsufficientFunds)` on the client.** The wire-deserialised value isn't structurally identical to the server's class. Always check `_tag`.
- **The curried `Schema.TaggedError('Name')({...})` form.** That doesn't type-check — use the class-extension form `class X extends Schema.TaggedError<X>()('X', {...})`.



---

<!-- source: en/data/cms.md -->
## CMS

_Headless CMS on the data layer — content types declared in code, derived draft/published tables, the ctx.cms read surface, and the saveDraft → publish write pipeline with save-time validation + derivation._

`@voltro/cms` is a headless CMS built on the framework's own data layer. You
declare a content type once, in code; the package compiles it to two real
database tables (`<type>_drafts` + `<type>_published`) that auto-migrate picks
up, gives you a typed read surface, and a write pipeline that enforces the
schema's validation/derivation rules on every save.

Two entry points keep the browser/server boundary clean:

- `@voltro/cms` — the server entry: table derivation (`@voltro/database`),
  the write pipeline, signed preview tokens (`node:crypto`), the REST mount.
- `@voltro/cms/web` — browser-safe: the field DSL, the validation/derivation
  engine + its typed errors (effect-only), and the `<ContentForm>` renderer
  (react-only). An editor page pre-validates with the SAME rules the server
  applies; a mutation descriptor can declare `error: ContentValidationFailed`
  without dragging the server graph into the bundle.

## Declaring a content type

By convention each content type lives in a `*.contentType.ts` file. The field
DSL is a plain-data descriptor language — validation rules travel as data and
are applied at save time.

```ts
import { defineContentType, Schema, derivedFrom, slugify } from '@voltro/cms'

export const blogPost = defineContentType({
  name: 'blogPost',
  displayName: 'Blog Post',
  pluralName: 'Blog Posts',
  fields: {
    title:       Schema.String.pipe(Schema.maxLength(200)),
    slug:        Schema.String.pipe(Schema.pattern(/^[a-z0-9-]+$/), Schema.unique(), derivedFrom('title', slugify)),
    body:        Schema.RichText({ allowImages: true, allowEmbeds: false }),
    tags:        Schema.Array(Schema.String),
    publishedAt: Schema.DateTime.optional(),
    coverImage:  Schema.Media().optional(),
    author:      Schema.Reference('author'),
    visibility:  Schema.Literal('public', 'unlisted', 'private'),
  },
  list: { columns: ['title', 'updatedAt'], defaultSort: { field: 'updatedAt', dir: 'desc' } },
})
```

- `maxLength` / `pattern` are enforced by the save-time validator (and
  `maxLength` also caps the shipped text widget).
- `derivedFrom(source, transform)` marks a field computed-on-save: the
  pipeline computes it from its source and **overwrites** a caller-provided
  value — `slugify` is the canonical transform.
- `unique()` marks a string field unique **per tenant** — the derived tables
  get a real composite `(tenantId, field)` DB UNIQUE constraint (the database
  rejects a duplicate slug within a tenant, not the app). Pair it with
  `derivedFrom('title', slugify)` for a unique slug.
- `Schema.RichText(options?)` stores a JSON document (e.g. a TipTap doc);
  `allowImages` / `allowEmbeds: false` reject image/embed nodes (or `<img>` /
  `<iframe>` / `<embed>` markers in string values) at save time. The shipped
  default widget is a plain textarea placeholder — swap in a real editor via
  the `ContentForm` `widgets` override.
- `Schema.Media()` stores a storage object key as text (e.g. one issued by
  `@voltro/plugin-storage`); resolve it to a URL after a read with
  `resolveMedia` (below).
- `list` describes the consuming editor's list view; every referenced name is
  validated at definition time.
- Reserved field names (`defineContentType` throws): `id`, `status`,
  `tenantId`, `createdAt`, `updatedAt`, `createdBy`, `updatedBy`.

`contentTypeToEntities(blogPost)` returns the `{ draft, published }` tables —
register them in your `database/index.ts` so auto-migrate creates them. Both
carry the lifecycle `status` column (`draft` / `published` / `archived`), the
`tenant()` mixin (tenant-scoped reads/writes through the request store), and
are reactive (the default) so a publish wakes live subscriptions.

## Reading — `ctx.cms`

`cmsContext(store, types)` builds the read surface over a request-scoped
store; attach it to your `AppContext` and read it with `useCms(ctx)`. Reads
target `<type>_published` and exclude `archived` rows.

```ts
import { cmsContext, useCms } from '@voltro/cms'

const posts = await useCms(ctx)
  .contentType('blogPost')
  .where('publishedAt', '<=', new Date())
  .orderBy('publishedAt', 'desc')
  .limit(20)
  .all()
```

`createCmsClient({ store, types })` is the same surface without a request
handle (build-time / static-site generation). Signed preview tokens
(`previewToken` / `verifyPreviewToken` / `cms.preview`) let an app read the
draft row instead — see the package README for the token contract and the
API-key-gated REST mount (`handleCmsRest`).

## Writing — the pipeline

The lifecycle ops take the caller's store first. Hand them `ctx.store` and
every write rides the runtime's auto-scope spine: tenant + audit columns are
stamped, reads are tenant-scoped, ids come from the table's id scheme.

```ts
import { saveDraft, publish, unpublish, archive } from '@voltro/cms'

// derive → validate → write into blogPost_drafts (status: 'draft')
const draft = await saveDraft(ctx.store, blogPost, input)

// copy the draft into blogPost_published (same id), atomic per row
const row = await publish(ctx.store, blogPost, draft.id as string)

await unpublish(ctx.store, blogPost, draft.id as string) // remove the published copy, keep the draft
await archive(ctx.store, blogPost, draft.id as string)   // status 'archived' — reads exclude it
```

`saveDraft` applies the `derivedFrom` derivations first, then validates the
derived row — a rule on a derived field checks the value actually written.
Violations throw `ContentValidationFailed` carrying **every** per-field
violation (`{ field, rule, message }`), so an editor can annotate the whole
form in one pass. Re-saving with `row.id` updates the caller's own draft;
`publish` copies into the published table (insert on first publish, update
after) and marks the draft in-sync.

Tenant isolation is structural: a foreign id reads as absent through the
scoped store (`ContentNotFound`), a caller-pinned `id` that exists only in
another tenant is refused (`ContentIdConflict`), and a caller-provided
`tenantId` never reaches a write — the pipeline owns the lifecycle columns.

Handlers written in `Effect.gen` use the Effect-native siblings — the same
typed error instances on the error channel:

```ts
import { Effect } from 'effect'
import { saveDraftEffect, publishEffect } from '@voltro/cms'

const program = Effect.gen(function* () {
  const draft = yield* saveDraftEffect(ctx.store, blogPost, input)
  return yield* publishEffect(ctx.store, blogPost, draft.id as string)
}).pipe(
  Effect.catchTag('ContentValidationFailed', (e) => Effect.succeed({ invalid: e.violations })),
)
```

### ISR revalidation on publish

A content type can declare which ISR routes fall when its content is
published or unpublished — `publish()`/`unpublish()` fire
[`revalidatePath` / `revalidateTag`](/docs/routing/render-modes#on-demand-revalidation)
for each entry after the write commits, reaching every `voltro start`
replica:

```ts
import { defineContentType, Schema } from '@voltro/cms'

const blogPost = defineContentType({
  name: 'blogPost',
  displayName: 'Blog post',
  pluralName: 'Blog posts',
  fields: {
    title: Schema.String.pipe(Schema.maxLength(200)),
    body:  Schema.RichText({ allowImages: true, allowEmbeds: false }),
  },
  revalidate: {
    paths: ['/blog/[slug]', '/blog'],
    tags: ['blog'],
  },
})
```

You don't need this on postgres for the plain publish case: a route declaring
`cacheInvalidatesOn: ['blogPost_published']` is already dropped by CDC when
the published table changes. Declare `revalidate` for what CDC can't see —
non-postgres dialects, routes whose loaders read the content indirectly, or
tag fanout across several routes.

## The engine, standalone

The validation/derivation engine is pure and exported on its own (also on
`/web`) for custom write paths and client-side form validation:

```ts
import { validateContent, contentViolations, applyDerivations } from '@voltro/cms/web'

const derived = applyDerivations(blogPost, formValue) // derivedFrom fields computed
const violations = contentViolations(blogPost, derived) // pure — never throws
validateContent(blogPost, derived) // throws ContentValidationFailed on any violation
```

## Scheduled publishing

A draft can be marked to go live at a future time, then flipped live by a
periodic sweep. `schedulePublish` stamps the draft's `publishAt` (a draft-only
nullable column) without publishing now; `publishDue` publishes every draft
whose time has passed.

```ts
import { schedulePublish, publishDue } from '@voltro/cms'

// Mark a draft to publish later (does NOT publish now):
await schedulePublish(ctx.store, blogPost, draftId, new Date('2026-01-01T09:00:00Z'))

// Flip every due draft live (idempotent; returns the published ids):
const publishedIds = await publishDue(ctx.store, blogPost, new Date())
```

`schedulePublish` keys off a tenant-scoped read (a foreign/missing id throws
`ContentNotFound`) and leaves `status: 'draft'` until the row goes live.
`publishDue` selects `publishAt <= now AND status = 'draft'`, runs the ordinary
`publish` per row, and clears `publishAt` — idempotent, so a row is never
double-published. It sweeps whatever store it is handed: a request store covers
one tenant; a background/root store covers every tenant.

`@voltro/cms` does not ship the scheduler — you drive `publishDue` from a
`*.cron.tsx` schedule:

```tsx no-check
// apps/api/schedules/content-scheduler.cron.tsx
import { defineSchedule } from '@voltro/runtime'
import { publishDue } from '@voltro/cms'
import { blogPost } from '../blogPost.contentType'

export default defineSchedule({
  name: 'contentScheduler',
  cron: '* * * * *',
  timezone: 'UTC',
  // app.store has no request subject, so the sweep flips due drafts live
  // across ALL tenants.
  handler: async ({ app }) => {
    await publishDue(app.store, blogPost, new Date())
  },
})
```

## Media resolution

A `Schema.Media()` field stores a storage object **key**. Turning it into a
served URL needs the storage service (its `getUrl` / `mintUrl` are
Effect-returning and `mintUrl` needs the request `Subject`), so that context
lives in your app — not the core package. Rather than couple `@voltro/cms` to a
storage plugin, resolution is a typed hook: you supply a `MediaResolver`, and
`resolveMedia` swaps every media key on a row (including media nested in
`Array` / `Struct` fields) for the resolved URL.

```ts
import { resolveMedia, resolveMediaAll, type MediaResolver } from '@voltro/cms'
import { Effect } from 'effect'

const resolver: MediaResolver = (key) =>
  Effect.runPromise(storage.mintUrl(key, ctx.subject)).catch(() => null)

const post = await useCms(ctx).contentType('blogPost').where('slug', '=', 'hello').one()
const withUrls = post ? await resolveMedia(blogPost, post, resolver) : null
```

`resolveMedia` returns a new row (the input is not mutated); a key the resolver
returns `null` for is left as its stored key. `resolveMediaAll` maps a whole
list; `mediaFields(type)` lists the top-level media field names.

## Versioning content

`@voltro/cms` ships no parallel revision system — the derived tables are
ordinary database tables, so `@voltro/plugin-row-history` gives full row history
plus time-travel with no new machinery. The plugin records every table by default —
narrow it with `include:` (pass the derived table handles) or `exclude:` if you
only want content history — then read a timeline or restore a snapshot:

```ts no-check
import { rowHistoryPlugin, rowHistory, restoreAsOf } from '@voltro/plugin-row-history'

// Register in your app's plugin list:
rowHistoryPlugin({})

const timeline = await rowHistory(ctx.store, 'blogPost_published', postId, tenantId)
await restoreAsOf(ctx.store, 'blogPost_published', postId, tenantId, someEarlierDate)
```

## The editor form

`<ContentForm>` (from `/web`) renders one labelled widget per field, driven by
each field's editor hint — stateless and unstyled. Override any widget (e.g. a
TipTap-backed `richText`) via the `widgets` prop:

```tsx
import { ContentForm } from '@voltro/cms/web'

<ContentForm contentType={blogPost} value={row} onChange={setRow}
  widgets={{ richText: MyTipTapWidget }} />
```



---

<!-- source: en/data/connections.md -->
## Connections (credentials vault)

_defineConnection — a per-user, encrypted store for third-party credentials, with the OAuth handshake, refresh-before-use, a resolver for handlers and plugins, and a connect UI, from one declaration._

An app that acts on a third party **on behalf of a user** — Jira with that
user's token, Slack with that user's OAuth grant — needs five things, none of
them app-specific and all of them easy to get subtly wrong:

1. a per-user token table,
2. encryption, so the tokens are not sitting in the database in the clear,
3. an OAuth authorize/callback pair with anti-CSRF state,
4. an expiry check with a refresh, without stampeding on the refresh token,
5. a resolver the request path can ask for "the credential for **this** user".

`defineConnection` is that whole stack from one declaration.

## Declare a connection

One `*.connection.ts` file per connection, default-exported. The file is
**server-only** — it holds a client secret, and it is never part of the browser
bundle.

```ts
// api/connections/jira.connection.ts
import { defineConnection } from '@voltro/runtime'
import { serverEnv } from '../env/server'

export default defineConnection({
  id: 'jira',
  kind: 'oauth2',
  label: 'Jira',
  authorizeUrl: 'https://auth.atlassian.com/authorize',
  tokenUrl: 'https://auth.atlassian.com/oauth/token',
  clientId: serverEnv.JIRA_CLIENT_ID,
  clientSecret: serverEnv.JIRA_CLIENT_SECRET,
  scopes: ['read:jira-work', 'offline_access'],
  authorizeParams: { audience: 'api.atlassian.com', prompt: 'consent' },
  identify: async (tokens) => {
    const res = await fetch('https://api.atlassian.com/me', {
      headers: { authorization: `Bearer ${tokens.accessToken}` },
    })
    const me = await res.json() as { account_id: string, email: string }
    return { accountId: me.account_id, accountLabel: me.email }
  },
})
```

For a provider with no OAuth app — or where a personal access token is simply
the supported path — use `kind: 'pat'`:

```ts
// api/connections/github.connection.ts
import { defineConnection } from '@voltro/runtime'

export default defineConnection({
  id: 'github',
  kind: 'pat',
  label: 'GitHub',
  instructionsUrl: 'https://github.com/settings/tokens',
  validate: async (token) => {
    const res = await fetch('https://api.github.com/user', {
      headers: { authorization: `Bearer ${token}` },
    })
    if (!res.ok) throw new Error('GitHub rejected that token')
    const me = await res.json() as { login: string }
    return { accountLabel: me.login }
  },
})
```

`validate` is worth writing. Without it, a mistyped token is accepted happily
and fails hours later inside a background job; with it, the user finds out while
they still have the token on their clipboard.

## Use the credential

`ctx.connections.get(id)` returns the **calling subject's** credential,
refreshed if it was near expiry. There is no parameter for "some other
subject" — the facade is bound to the request.

```ts
// api/actions/syncIssues.action.server.ts
export default async function syncIssues(input: { projectKey: string }, ctx) {
  const jira = await ctx.connections.get('jira')

  const res = await fetch(`https://api.atlassian.com/ex/jira/search?jql=project=${input.projectKey}`, {
    headers: { authorization: `Bearer ${jira.accessToken}` },
  })
  return { count: (await res.json()).total }
}
```

Use `ctx.connections.tryGet(id)` for the "use it if we have it" branch — it
returns `null` when nothing is connected. A failing *refresh* still throws:
silently degrading a connected-but-broken account to "not connected" would hide
a revoked grant behind a feature that quietly does nothing.

`ctx.connections` is **absent** when the app declares no `*.connection.ts`, so
reaching for it in an app with no connections is a type error rather than a
runtime surprise.

## The connect UI

`useConnection` gives you one connection's live state plus its operations.
Everything reactive comes from a single subscription, so a completed OAuth
callback updates an open settings page with no polling.

```tsx
import { useConnection } from '@voltro/client'
import { ConnectAccount } from '@voltro/ui'

export const IntegrationSettings = () => {
  const jira = useConnection('app', 'jira')
  const github = useConnection('app', 'github')

  return (
    <>
      <ConnectAccount connection={jira} loading={jira.loading} />
      <ConnectAccount connection={github} instructionsUrl="https://github.com/settings/tokens" />
    </>
  )
}
```

`useConnections('app')` returns the whole list if you would rather render the
page yourself:

```tsx
const { connections, get, loading } = useConnections('app')
// each entry: { connectionId, kind, label, status, accountLabel, scopes, expiresAt, lastError }
```

`connect()` opens the provider's consent screen in a popup by default (which
keeps the current page — and any unsaved form state — mounted) and falls back to
a full navigation when the popup is blocked. Pass `{ mode: 'redirect' }` to
navigate the tab instead.

**The token is never on the client.** The wire shape carries status, account and
expiry — nothing else. If you find yourself wanting the token in the browser,
the call that needs it belongs on the server.

## Status

| Status | Meaning |
| --- | --- |
| `disconnected` | No credential on file for this user. |
| `connected` | A usable credential is on file. |
| `expired` | The access token is past its deadline and there is no refresh token. Re-consent needed. |
| `revoked` | The provider **refused** the refresh. The stored tokens were cleared; only reconnecting restores it. |
| `error` | The last refresh failed transiently (5xx / network). Tokens retained; the next use retries. |

`expired`/`revoked`/`error` are deliberately not "connected" — that distinction
is what makes the UI prompt rather than silently do nothing.

## Encryption is not optional

Tokens are written through the framework's field cipher — the same
`FieldCipher` `.encrypted()` columns use, not a second mechanism with its own
key management. **An app that declares a connection and has no cipher
configured refuses to boot**, with a message naming the connections. There is no
plaintext fallback and no warn-and-continue path.

```ts
// app.config.ts
import { governancePlugin } from '@voltro/plugin-governance'

export default defineApiConfig({
  plugins: [
    governancePlugin({ fieldEncryption: { secretKey: 'VOLTRO_FIELD_ENCRYPTION_KEY' } }),
  ],
})
```

## Refresh

Resolving checks the deadline with 60s of skew and renews **ahead** of it, so no
call site needs a retry-on-401 dance. Two layers keep concurrent refreshes from
stampeding:

- an in-process single-flight, so many handlers on one replica share one
  refresh;
- a compare-and-set lease on the row, so two **replicas** do not both spend the
  refresh token — which, with a provider that rotates refresh tokens, would
  invalidate the grant outright.

A provider that does not rotate simply omits `refresh_token` from its reply; the
existing one is kept rather than overwritten with null.

## Plugins

A plugin never sees a request context, so the vault also publishes a
process-level resolver. `@voltro/plugin-atlassian` consumes it directly:

```ts
// app.config.ts
import { atlassianPlugin } from '@voltro/plugin-atlassian'
import { connectionCredentials } from '@voltro/plugin-atlassian/connection'

plugins: [
  atlassianPlugin({
    credentialsResolver: connectionCredentials({
      connectionId: 'jira',
      baseUrl: serverEnv.JIRA_BASE_URL,
    }),
  }),
]
```

The plugin's contract is unchanged — it still receives a
`(subject) => Effect<AtlassianCredentials>` and still never reads your schema.
What changes is who owns the token. A plugin with its own credential resolver
keeps working exactly as before; this is an additional way to satisfy the same
option, not a replacement.

## What the framework stores

Two tables, created only when the app declares a connection:

- `_voltro_connections` — one row per (connection, subject). Access and refresh
  tokens are ciphertext. Reactive, which is what makes the connect UI live.
- `_voltro_connection_grants` — an in-flight OAuth handshake (single-use state,
  encrypted PKCE verifier, 10-minute TTL).

The callback endpoint is `GET /_voltro/connections/<id>/callback`, mounted
automatically under both `voltro dev` and `voltro serve`. Register it as the
redirect URI on your OAuth app, or set `redirectUri` on the declaration. The
app's public origin comes from `VOLTRO_PUBLIC_URL`.

## Deliberate non-goals

- **Disconnect does not revoke at the provider.** Providers disagree on whether
  a revocation endpoint exists, what it takes, and whether it kills sibling
  sessions. `disconnect()` forgets our copy and says so, rather than pretending
  to a revocation it cannot guarantee. Revoke in the provider's own UI when that
  matters.
- **No app-wide connection.** A credential belongs to a subject. An anonymous
  caller is refused rather than bucketed under a shared pseudo-subject, which
  would be a credential every visitor shares. For a service account, store it as
  a secret, not a connection.
- **`redirectTo` is a same-origin path only.** An absolute URL is rejected —
  otherwise every app declaring a connection would ship an open redirector.