Change streams

A change stream is a long-lived HTTP connection that receives one server-sent event per committed mutation on an entity. The engine ships the whole path — an event hook, an in-process subscription manager, and a server-sent-events transport — but only the engine library’s own REST router mounts the route.

Where this is wired

Read this table before anything else on the page. It decides whether you can use change streams at all.

Component	Where it lives	State
Event hook (publishes on create/update/delete)	Engine hook chain	Registered in data.svc, with no bus — it returns immediately and emits nothing
Subscription manager (topics, filters, fan-out)	Engine library	Implemented; constructed only by an embedding host
Server-sent-events transport	The library’s own REST router	Implemented; mounted only when that router is used
GET {prefix}/subscribe/{entity}	The library’s own REST router	Not served by data.svc
subscriptions: YAML block	Schema loader	Entity-level: parsed and stored, never read. Schema-level: parsed, then dropped at conversion

data.svc serves no change-stream route

The deployable service mounts these surfaces and no others: schema, REST, GraphQL, bulk, query-hash, actions, scripted endpoints, stateflows, admin and superadmin. There is no subscribe route among them, and the event hook it registers has no bus attached, so an events: block on an entity publishes nothing. Change streams are available only to a Go host that embeds the engine, wires a manager, and mounts the library router. There is no GraphQL subscription fallback either — see GraphQL.

The route

One route is registered per entity in the schema, at router construction:

GET {prefix}/subscribe/{entity}

{prefix} is the library router’s URL prefix, /api unless the host overrides it. The entity name is captured at registration, so a request to an entity that is not in the schema is a plain 404 from the mux.

Status	Cause
200	Stream opened; frames follow until either side disconnects
404	No route for that entity name
500	The response writer does not support flushing (streaming not supported by the server)
503	No identity function is wired on the router (subscriptions require server identity wiring)

The 503 is the cross-tenant guard. Identity supplies the tenant that scopes the subscriber’s filter; without it every subscriber would be unscoped, so the handler refuses to open the stream rather than serving a firehose.

Query parameters

Parameter	Values	Default	Behaviour
ops	comma-separated create, read, update, delete	all operations	Bitmask filter on the event’s operation
heartbeat	Go duration string (45s, 1m) or bare integer seconds	30s	Keep-alive comment interval

A misspelled ops value widens the stream instead of narrowing it

Unknown tokens are dropped silently. ?ops=upsert parses to an empty mask, and an empty mask means no constraint — you get every operation rather than none. Check the tokens; there is no error for a typo. ?ops=read is legal but matches nothing: the publishing hook subscribes to writes only, so no read event is ever produced.

An unparseable heartbeat value falls back to 30s without an error. The default sits under the common reverse-proxy idle timeouts (60s on nginx, 120s on an ALB), so the connection stays open through an idle period.

Response headers

Header	Value
Content-Type	text/event-stream
Cache-Control	no-cache
Connection	keep-alive
X-Accel-Buffering	no

X-Accel-Buffering: no is set because nginx and CloudFront buffer text/event-stream by default, which would hold frames until a buffer fills.

Frames

The handler writes a comment frame on connect, then one frame per delivered event, plus a heartbeat comment on each tick.

: connected sub=17

id: evt-1749728461000000000
event: entity.create
data: {"id":"evt-1749728461000000000","type":"entity.create","source":"orders-api/orders","subject":"01J8...","time":"2026-07-26T10:41:01Z","entity":"orders","operation":"create","tenant":"acme","user_id":"u-42","data":{"id":"01J8...","status":"draft","total":1200}}

: ping

: server shutting down

sub=<n> is the monotonic subscriber id, useful for correlating a connection with its counters in logs. id: and event: are emitted so a browser EventSource can dispatch by type and track the last event id. Payloads are JSON-encoded on one line — the event-stream format splits on newlines, so a pretty-printed body would emit continuation lines.

The stream ends on any of: client disconnect, a failed write to the client, or manager shutdown (preceded by the : server shutting down comment). Every exit path unsubscribes and releases the buffer.

Event envelope

Field	Type	Content
id	string	Event id. Defaults to evt-<unix nanoseconds> unless the host supplies a generator
type	string	entity.create, entity.update or entity.delete
source	string	<source>/<entity>, where the source prefix is set by the host
subject	string	Primary-key value, when the payload carries one; omitted otherwise
time	RFC 3339	Publication time, UTC
entity	string	Entity name
operation	string	create, update or delete
tenant	string	Tenant of the mutation
user_id	string	User who performed the mutation
data	object	Write payload after the entity’s include: / exclude: filtering

data is the write payload, not a protected read

The payload is captured after validation and before SQL, and it is shaped only by the event config’s include: / exclude: lists. It does not pass the read path, so field protection and row-level security do not shape it. A field you would mask on read is delivered in cleartext on the stream unless you exclude it in the event config. Tenant scoping is the one isolation the transport enforces itself.

The type values are the operation tokens, not past-participles. The AsyncAPI document produced by schema export describes channels named entity.<name>.created / .updated / .deleted; those channel names do not correspond to any topic or event: value this transport uses.

Topics

Events are published to a topic string. The default topic for an entity is entity.<name>, and the subscribe handler subscribes to exactly entity.<entity-from-the-path>.

Topic	Who receives it
entity.<name>	Subscribers on that topic, plus wildcard subscribers
*	Every publication, on any topic — Go subscribers only, no route serves it

A custom topic: is unreachable over the stream

An entity’s events[].topic overrides the publish topic, but the subscribe handler builds entity.<name> from the URL path and cannot be told otherwise. Set a custom topic and the stream for that entity stays empty. Leave topic: unset for anything you intend to consume over server-sent events.

Filters

A subscriber carries a filter applied at fan-out, after topic matching. Structural dimensions are checked first; the script predicate runs only if they pass.

Dimension	Semantics	How the stream sets it
Entity	Case-insensitive entity-name match; empty means any	From the URL path
Operations	Operation bitmask (create 1, read 2, update 4, delete 8); 0 means no constraint	From ?ops=
TenantID	Exact match; empty means every tenant	From the router’s identity function
UserID	Exact match on the mutating user; empty means every user	Always empty — the transport does not expose it
Script	Script predicate; truthy delivers, falsy or an error drops	Not exposed by the transport

The script predicate receives event, entity, operation, data, tenant and user_id, and is evaluated only when the host installed a script-evaluation callback on the manager. Without one, script filters are treated as always-allow — they fail open, not closed.

Delivery contract

Each connection gets a channel of 64 events. Publication is non-blocking: the publisher walks the matching subscribers and enqueues without waiting for any of them to drain.

Condition	Result
Buffer has room	Event enqueued; the subscriber’s delivered counter increments
Buffer full	Event dropped for that subscriber only; its dropped counter increments; other subscribers are unaffected
Subscriber already unsubscribed	Skipped before the send
Filter rejects the event	Skipped; counted as neither delivered nor dropped

A slow consumer therefore loses events; it never slows the mutation that produced them, and it never blocks another subscriber. A non-zero drop counter is the signal that a consumer cannot keep up.

There is no replay and no gap detection

id: is emitted, but the handler does not read Last-Event-ID and the manager keeps no history. A reconnecting client resumes live and never learns what it missed — whether the gap came from a full buffer or from being disconnected. Treat the stream as a notification that something changed and re-read the entity over REST when you need the authoritative row.

Two further paths produce no events at all:

• Bulk create. The bulk insert path runs only the two validation phases per record, never the post-execution phase where publication happens. Rows inserted in bulk are silent.
• Entities with no events: block. The publishing hook returns immediately when the entity declares no event configs. Declaring the entity is not enough; declare the event.

Counters

Scope	Counters
Per subscriber	delivered, dropped
Per manager	active subscribers, active topics, published, delivered, dropped (cumulative)

Both are read through the Go API. No HTTP endpoint publishes them, so a host that wants them in telemetry has to export them itself.

Publishing side

An entity publishes only what it declares. The event hook runs after the mutation statement returns, so a statement that errored publishes nothing.

entities:
  - name: orders
    events:
      - name: order_changed
        operations: [create, update]
        exclude: [internal_notes, card_last4]
        condition:
          language: expr
          expression: payload.status != "draft"

Key	Type	Default	Meaning
name	string	—	Label; also used in the publish-error metadata key
operations	list	all write operations	create, update, delete, matched case-insensitively
condition	script ref	—	Guard; a falsy result skips this event. An evaluation error aborts the mutation
topic	string	entity.<name>	Publish topic
format	string	—	cloudevents, json or avro. Stored; the transport always writes JSON
include	list	—	Allow-list of payload fields kept in data
exclude	list	—	Deny-list of payload fields removed from data. Ignored when include is set

A publish failure is recorded in the request metadata and does not fail the mutation. A guard evaluation failure does — the condition script is on the request path.

The event fires before commit

The hook runs at the after-execution phase, which is inside an open transaction when the mutation is part of one. A transaction that rolls back after the statement succeeded has already published its events. An event is evidence the statement ran, not that it committed.

The bundled examples use a different dialect

The example schemas in the repository write events: in an older dialect: trigger: update instead of operations: [update], a bare string for condition:, and publish: nesting topic:, format: and payload.include. The deployable loader reads none of those. A scalar condition: is a type error that fails the whole file to load; trigger: and publish: are unknown keys that are dropped in silence — which leaves operations empty (matching every write) and discards the include allow-list, so the full payload ships. Rewrite copied blocks against the table above.

The subscriptions: block

The loader accepts a subscriptions: list in two places, and treats them differently. An entity-level list is converted and stored on the entity. A schema-level list is parsed and merged by name across the files of a multi-file schema, then discarded — the typed schema has no schema-level subscription field, so nothing survives conversion. The two levels are never merged with each other.

Neither one is read at runtime: no transport is selected from a declaration, no filter is built from it at boot, no scope is enforced from it.

Key	Values
name	string
source	topic name
eventtypes	create, update, delete
scope	global, tenant, user
rls	bool
fieldgroup	field-group name
transports	sse, websocket, graphql, grpc
filter	script ref

It is documented here so you recognise it in an inherited schema. Do not model behaviour on it. The filter that actually applies is the one the transport builds per connection, described above.

Mounting the transport

An embedding host constructs the manager, gives it to the event hook as its bus, and passes it to the router alongside an identity function. Both are required: the manager without the router publishes to nobody, and the router without an identity function answers 503.

mgr := subscription.NewManager(0)
mgr.SetScriptEval(evalScript) // optional; without it script filters always pass

// Register on the engine's hook chain. Bus is what makes the hook do anything.
evt := &hook.EventHook{
    Bus:      mgr,
    Source:   "orders-api", // becomes "orders-api/<entity>" in event.source
    EvalBool: evalBool,     // evaluates events[].condition
}

router := rest.NewRouter(engine, schema,
    rest.WithPrefix("/api"),
    rest.WithIdentity(identityFromToken), // returns tenant, user, roles
    rest.WithSubscriptions(mgr),
)

Set Source explicitly. Unset, it falls back to an internal placeholder that will appear in every consumer’s source field.

Call the manager’s Close at shutdown: it terminates every subscriber, which makes each open stream write its shutdown comment and exit rather than hanging until the client times out.

Consume the subscriber handle by selecting on both its event channel and its done channel. The event channel is never closed — ranging over it alone hangs forever after unsubscribe.

Related

For	See
The events:, triggers: and pointcuts: keys in context	Entities
Who may read the entity at all	Access rules
The AsyncAPI contract document	Schema export
Every served route, parameter and status code	Data API reference