Live Queries

A subscription (previous page) 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 (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.

The request01

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). 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 stream02

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

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.

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 delta03

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):

Exact top-N: the cushion + Postgres oracle04

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: LiveSpec05

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

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 bridge06

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 the scope-subscription plane uses.

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)
}

Authorization07

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

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.

Performance08

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):

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 architecture09

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 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 shards10

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 tier11

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 next12