---
title: Live Queries
description: Maintained result sets — filter + sort + limit subscriptions that emit exact enter/leave/move/update deltas, with Postgres as the ordering oracle.
---

A *subscription* ([previous page](/docs/fabriq/(concepts)/subscriptions)) tells you that some row in a declared scope changed. A **live query** goes further: you hand fabriq a filter, a sort, and a window size, and it **maintains the result set for you**, emitting `enter` / `leave` / `move` / `update` deltas as the world changes — RethinkDB-changefeed / Convex / Meteor / Firestore live-query semantics. The visible window is always the true first-N of the total order: **exact top-N at all times**.

Live queries reuse the structured filter from the [relational plane](/docs/fabriq/(data-planes)/relational) (`query.Where`) and keep **Postgres authoritative** for ordering and the window boundary, so the in-engine fast path never has to reproduce SQL collation semantics to stay correct.

<Callout type="info">
  Live queries are **additive**. Scope subscriptions (`f.Subscribe`) remain the right tool for "tell me when anything in this scope changes"; reach for a live query when the client wants a *maintained, ordered, filtered window* it can bind directly to a list UI.
</Callout>

## The request

```go
type LiveQuery struct {
	Entity string        // registry entity name
	Where  query.Where   // the same filter AST as relational List (Eq/In/Gt/Like/Or/IsNull…)
	Sort   []SortKey     // ordered; fabriq appends id ASC as the final unique tiebreak
	Limit  int           // window size N
	Cursor *Cursor       // keyset anchor (sort-key tuple); nil = from the head
	Mode   Mode          // ModeMaintained (default)
}

type SortKey struct { Column string; Desc bool }
```

The filter columns are validated against the entity (the injection guard, via `query.ValidateConds`), and the sort columns against the entity's declared **sortable** set (see [`LiveSpec`](#opting-in-livespec)). Fabriq always appends `id ASC` as the final tiebreak, so `(Sort…, id)` is a **total order** — required for stable keyset pagination. There is deliberately **no offset**: offset pagination is unstable under churn, so live windows use keyset cursors only.

## Two phases over one stream

`LiveQuery` returns an initial **snapshot** plus a channel of live **deltas**:

```go
func (f *Fabriq) LiveQuery(ctx context.Context, q livequery.LiveQuery) (
	livequery.Snapshot, <-chan livequery.LiveDelta, *livequery.Handle, error)
```

The third return is a `*livequery.Handle`: call `Close()` to tear the subscription down, and `Reanchor(ctx, *Cursor, limit)` to refill a fresh window when deep-scrolling.

```go
snap, deltas, handle, err := f.LiveQuery(tenantCtx, livequery.LiveQuery{
	Entity: "asset",
	Where:  query.Where{query.Eq("kind", "pump"), query.Eq("site_id", siteID)},
	Sort:   []livequery.SortKey{{Column: "name"}},
	Limit:  50,
})
if err != nil { /* unknown entity, no LiveSpec, bad column, denied authz, not configured */ }
defer handle.Close()

render(snap.Rows)            // the initial ordered window, straight from Postgres
for d := range deltas {       // then splice each delta into the rendered list
	apply(d)
}
```

The **snapshot** is read through the relational path (RLS-enforced) and carries the first `N` rows in total order. The handoff is **gapless**: the engine attaches to the live change feed *before* taking the snapshot and discards any change whose version the snapshot already reflects, so no row is missed or double-applied across the seam.

## The delta

```go
type LiveDelta struct {
	Op       DeltaOp         // OpEnter | OpLeave | OpMove | OpUpdate | OpReset
	AggID    string
	Version  int64           // idempotency key with AggID; stale versions are discarded
	Row      json.RawMessage // column-keyed payload (enter/update/move)
	OldIndex int             // window position before (leave/move/update); -1 if N/A
	NewIndex int             // window position after  (enter/move/update); -1 on leave
	Cursor   Cursor          // stable sort-key tuple — for array splicing AND resume
	StreamID string          // → SSE id:, drives Last-Event-ID resume
	At       time.Time
}
```

Each delta carries **both** a numeric window index (for array-splicing list UIs) and the stable sort-key `Cursor` (for resume and correctness):

| Op | Meaning | Client action |
|----|---------|---------------|
| `enter` | row entered the visible window | insert at `NewIndex` (may push the last row out) |
| `leave` | row left the visible window | remove at `OldIndex` |
| `move` | row stayed visible, position changed | move `OldIndex` → `NewIndex` |
| `update` | row stayed at its position, payload changed | replace at `NewIndex` |
| `reset` | discard the window and re-snapshot | refetch (emitted on failover / overflow) |

## Exact top-N: the cushion + Postgres oracle

The maintenance contract is *exact top-N at all times*. The engine holds an ordered prefix of the result — the visible window `[0, N)` **plus a cushion** `[N, N+C)` — maintained as a **true prefix of the Postgres-ordered result**:

> **Invariant:** the buffered rows are exactly the first `len(rows)` of the Postgres-ordered result from the anchor. Postgres owns ordering; the in-memory fast path only splices.

On each change the engine evaluates the filter in Go (fast path) and decides where the row sorts relative to the buffer. When a row **leaves** the visible window, the first cushion row is promoted to take its slot; when the cushion runs low, a single bounded **keyset refill** (`WHERE (sort…, id) > cursor ORDER BY … LIMIT k`) tops it back up from Postgres. Because Postgres — not hand-rolled Go comparison — fills and orders the buffer, text collation and type-coercion subtleties can never corrupt the window. This is the *hybrid* design: an in-engine incremental matcher on the hot path, Postgres authoritative at exactly the two moments correctness is hard (snapshot and boundary).

## Opting in: `LiveSpec`

An entity opts into live queries by declaring a `LiveSpec` (nil = disabled), mirroring how `SearchSpec` opts into the search plane:

```go
r.MustRegister(registry.EntitySpec{
	Name:  "asset",
	Model: (*domain.Asset)(nil),
	Live: &registry.LiveSpec{
		Filterable: []string{"name", "kind", "site_id"}, // columns allowed in Where (empty = all)
		Sortable:   []string{"name", "kind"},            // columns allowed in Sort  (empty = all)
		MaxWindow:  500,                                  // cap on Limit
	},
})
```

Columns are validated against the model at registration — an unknown filterable/sortable column fails fast.

## The SSE bridge

Because the query body (filter + sort + limit) does not fit a query string, the subscribe is a `POST` that upgrades to a Server-Sent Events stream. The example service exposes it at `POST /api/v1/live`: it writes the snapshot as a `snapshot` event, then each delta as an event named for its op (`enter`/`leave`/`move`/`update`) with `id` = `StreamID` for Last-Event-ID resume, over the same proxy-safe [`SSEWriter`](/docs/fabriq/(concepts)/subscriptions#the-sse-bridge) the scope-subscription plane uses.

```go
snap, deltas, cancel, err := s.fabric.LiveQuery(tctx, q)
// ...
_ = sse.WriteEvent("", "snapshot", snap)
for d := range deltas {
	_ = sse.WriteEvent(d.StreamID, d.Op.String(), d)
}
```

## Authorization

A `WithLiveAuthz` hook runs before the snapshot, receiving the tenant-stamped context and the query:

```go
fabriq.Open(ctx, reg, cfg, fabriq.WithLiveAuthz(func(ctx context.Context, q livequery.LiveQuery) error {
	// allow/deny; later phases may also inject mandatory row-visibility predicates into q.Where
	return nil
}))
```

Tenancy is always structural — the tenant comes from the authenticated context and is never client-supplied, and snapshots/refills run under the RLS-scoped app role.

## Performance

The package ships microbenchmarks for the hot path (`go test ./core/livequery/ -bench=. -benchmem`). Representative numbers (Apple M-series, illustrative — run them on your own hardware):

| Benchmark | What it measures | Cost |
|-----------|------------------|------|
| `PredicateEval` | one filter match against a row | ~80 ns, 0 allocs |
| `WindowApplyUpdate/N=10` | per-event maintenance, 10-row window | ~120 ns |
| `WindowApplyUpdate/N=1000` | per-event maintenance, 1000-row window | ~940 ns |
| `WindowApplyChurn` | full enter/evict ↔ leave/promote cycle | ~1.2 µs |
| `WindowFanout/subs=1000` | routing one change to 1000 subscriptions | ~140 µs |
| `WindowFanout/subs=10000` | routing one change to 10000 subscriptions | ~1.5 ms |

Two things to read from these. Per-subscription maintenance is cheap and grows with the **window size** (the membership lookup), not the result-set size — deep windows stay affordable. The `WindowFanout` row measures the cost of applying one change to N windows *unconditionally*; the engine no longer does that — it routes each change only to the subscriptions a content-based predicate index identifies as possible matches (plus current holders, to catch leaves), so a change touches `O(candidates)`, not `O(subscriptions)`.

## Scaling architecture

Subscriptions to the same `(tenant, entity)` share one **partition**: one feed, one predicate index, and one lock-free dispatch goroutine (an actor — all state owned by a single goroutine, so it is race-free by construction). On each change the dispatcher selects candidates from the predicate index (the [counting algorithm](https://en.wikipedia.org/wiki/Content-based_publish/subscribe) over conjunctive equality constraints) unioned with the change's current holders, and folds it into only those subscriptions.

Within a partition, **identical query shapes share one view** — one window, one matcher — keyed by shape. A saved view watched by a thousand clients costs one window; the first subscriber seeds it from Postgres and the rest attach to its live state, with deltas fanned out. This is *templated sharing*, and it falls out of the view abstraction for free.

A subscription chooses its delivery mode:

- **Maintained** (default) — the exact ordered window described above. `Subscribe` returns a `*Handle` whose **`Reanchor(cursor, limit)`** slides the window to a new anchor for deep/infinite scroll, re-keying onto the new shape's view at `O(window)` server cost.
- **Streamed** (`ModeStreamed`) — holds a compact membership **ID-set** (seeded from a `MemberLister`) instead of an ordered buffer, and forwards `+match` / `−unmatch` / `update` transitions for the client to order. No boundary refill, no per-row buffer — the variant that scales to enormous or high-cardinality result sets.

A **reconcile backstop** (`f.ReconcileLiveQueries(ctx)`, run on a low cadence by the worker) re-checks each maintained view against Postgres truth and re-snapshots any that drifted (emitting `OpReset`), so transient predicate-evaluation divergence self-heals.

## Scaling out across shards

The engine above also runs sharded across many processes (`core/livequery/cluster`). Data maps to one of `Partitions` (256) buckets by `hash(tenant, entity)`; the relay publishes each event to that bucket's stream `lq:events:{p}`. Live shards divide the buckets among themselves by **rendezvous (HRW) hashing** over heartbeat-registered membership — coordinator-free, so every node computes the same `Owner(p)` independently and removing a shard reassigns only *its* partitions.

A **`Shard`** is the single-node `Engine` reused unchanged, wired to its partitions' streams and gated by ownership. It heartbeats, serves `subscribe`/`reanchor`/`unsubscribe` control routed to it over `lq:ctrl:{shard}`, and pumps deltas back to gateways over `lq:delta:{gateway}`. A **`Gateway`** terminates the client, routes control to the owning shard, and demuxes its delta channel.

**Failover is transparent.** When a shard dies its heartbeat expires, `Owner(p)` shifts, and the new owner — detecting the change — rebuilds the partition's subscriptions from the durable registry (`ByPartitionNum`) and re-snapshots them; the client sees an `OpReset` on the same stream and keeps flowing. Reanchored scroll positions survive too (the registry stores the live cursor). This is proven end-to-end by a multi-process harness (shards as goroutines over real Redis/Postgres) that kills the owning shard mid-stream.

## The gateway tier

The in-process `Gateway` also ships as a **deployable edge tier** (`gateway` package + a Forge extension): it terminates client **SSE** and **WebSocket** connections and forwards the maintained/streamed delta stream over them, layered on the same control/delta protocol — no new live-query logic, just transport.

- **SSE** — `POST /api/v1/live` with a `LiveQuery` body returns a uniform event stream: the snapshot folded in as `reset`+`enter` events, then live `enter`/`leave`/`move`/`update`. A single, stateless endpoint — *reanchor* is a reconnect with a new cursor (a fresh snapshot at the new anchor), *unsubscribe* is a disconnect.
- **WebSocket** — `GET /api/v1/live/ws` is bidirectional: deltas stream down as frames while the client sends `subscribe`/`reanchor`/`unsubscribe` commands up. Forge owns the upgrade (the gateway carries no third-party WebSocket dependency).

**Reconnection is re-subscribe + fresh snapshot:** the gateway holds no cross-connection state, so a dropped client simply re-sends its query and gets a new `OpReset`+snapshot — the same code path as failover, and a reconnect may land on any gateway instance. **Backpressure** tears a hopelessly slow client down (SSE via a write deadline, WebSocket via a write watchdog) and lets it reconnect to a fresh snapshot, so a slow consumer never consumes unbounded memory or silently diverges.

It is packaged as a Forge extension that builds the `Gateway` over the facade's Redis transport and registers the two controllers; auth and AsyncAPI/OpenAPI schemas are caller-supplied route options, so the gateway stays auth-scheme-agnostic. The controllers are exported, so a host app can also mount them on its own router directly.

## Where to go next

<Cards>
  <Card title="Subscriptions" href="/docs/fabriq/(concepts)/subscriptions">The scope-based delta plane live queries build on.</Card>
  <Card title="Relational" href="/docs/fabriq/(data-planes)/relational">The `query.Where` filter and List path live queries reuse.</Card>
  <Card title="Registry" href="/docs/fabriq/(concepts)/registry">How `LiveSpec` and subscription scopes are declared per entity.</Card>
</Cards>