Realtime

Subscribe

A subscription is a live version of a query: the same filter, sort and limit, but instead of one response you get a snapshot followed by a stream of windowed deltas over Server-Sent Events. The engine holds the window for you — no client-side diffing, buffering, or version bookkeeping.

Endpoints

The body is the same one …/query takes, plus two stream-only knobs:

{
  "objectId": "obj_abc",
  "dataset": "chat_messages",
  "filter": { "creator": "A5k…" },
  "sort": ["-_ver.id"],
  "limit": 50,
  "offset": 0,
  "includeTotal": true,
  "mailboxCapacity": 256,
  "driftBudgetPercent": 30
}

The field set is closed: an unknown top-level key answers 400 request.unknown_field. Errors raised before the stream opens (space.not_found, request.missing_field, …) come back as the normal JSON error envelope; once the response is 200 text/event-stream, problems are closed frames.

Frames

event: ready
data: {}

event: snapshot
data: {"records":[{"id":"msg_1","text":"hello","_ver":{…}}],"total":17,"hasNext":true}

event: changes
data: [{"versionId":"!!%>",
        "added":  [{"id":"msg_2","doc":{…},"ops":[{"type":"$set","path":[],"payload":{"text":"hi"}}]}],
        "updated":[{"id":"msg_1","doc":{…},"ops":[{"type":"$set","path":["text"],"payload":"hello!"}]}],
        "removed":[{"id":"msg_0","reason":"displaced"}]}]

: keepalive

event: closed
data: {"reason":"overflow"}

1. ready — sent once the subscription is registered. Wait for it before treating the stream as live.
2. snapshot — sent once, right after ready. records is the materialized window; total / hasNext appear only with includeTotal. The snapshot and the first event sit at adjacent versions with nothing missed in between, so integrate the snapshot first, then apply changes in order.
3. changes — a JSON array of one or more events, each {versionId, added, updated, removed}. added and updated carry the full post-apply doc plus the $set / $unset ops of the triggering change. Most clients overwrite their local entry with doc; clients that want atomic field merges apply ops (path: [] with an object payload is a multi-field set at the record root). $inc, $addToSet and $pull are never emitted — the engine collapses them to the merged $set before delivery, so non-Go clients never reimplement CRDT merge rules.
4. : keepalive — a comment every ~25 s while idle, defeating idle proxy timeouts.
5. closed — terminal.

versionId is the per-change DAG order on this peer. It is useful for fencing ("processed up to X") but is not comparable across devices.

removed carries a reason

Only deleted means the object is gone. For the other two, a fresh snapshot would return the record again.

Closing and recovery

Recovery is the same for all four: open a new POST and take the fresh snapshot. There is no replay across reconnects and no resume cursor — the new snapshot already reflects current state, which is strictly cheaper than reconstructing it from a backlog. overflow and drifted are split only so you can log and back off sensibly; a burst of overflow on a hot collection is the hint to raise mailboxCapacity, a stream of drifted on a churny list is the hint to raise driftBudgetPercent or widen limit.

Note. Drift detection needs a window to measure against: with limit: 0 there is no window auto-shift and no drift safety net. Always subscribe with a limit.

Client example: fetch streaming, no EventSource

Windowed subscribes are POST, so the browser's EventSource cannot open them. Parse the SSE frames from a streaming fetch body instead:

async function subscribe(url, body, onFrame) {
  const res = await fetch(url, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(body),
  });
  if (!res.ok) throw new Error((await res.json()).error.code);

  const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
  let buf = "";
  for (;;) {
    const { value, done } = await reader.read();
    if (done) return;                       // connection dropped without `closed`
    buf += value;
    let i;
    while ((i = buf.indexOf("\n\n")) >= 0) {  // one frame per blank line
      const frame = buf.slice(0, i); buf = buf.slice(i + 2);
      let event = "message", data = "";
      for (const line of frame.split("\n")) {
        if (line.startsWith("event:")) event = line.slice(6).trim();
        else if (line.startsWith("data:")) data += line.slice(5).trim();
        // lines starting with ":" are keepalive comments — ignore
      }
      if (data) onFrame(event, JSON.parse(data));
    }
  }
}

const window = new Map();
function run() {
  subscribe("http://127.0.0.1:7001/v1/spaces/SPACE/query/subscribe",
    { objectId: "CHAT", dataset: "chat_messages", sort: ["-_ver.id"], limit: 50 },
    (event, data) => {
      if (event === "snapshot") for (const r of data.records) window.set(r.id, r);
      if (event === "changes") for (const ev of data) {
        for (const r of ev.added)   window.set(r.id, r.doc);
        for (const r of ev.updated) window.set(r.id, r.doc);
        for (const r of ev.removed) window.delete(r.id);
      }
      if (event === "closed") setTimeout(run, 500);   // every reason: resubscribe
    }).catch(() => setTimeout(run, 2000));
}
run();

From the shell, curl -N shows the raw frames, and the CLI prints one JSON object per frame ({"event": …, "data": …}) so the stream pipes through jq:

curl -N http://127.0.0.1:7001/v1/spaces/SPACE/objects/query/subscribe \
  -H 'Content-Type: application/json' \
  -d '{"filter":{"any.types":"page"},"sort":["-modifiedAt"],"limit":20}'

any query-subscribe SPACE --filter '{"any.types":"page"}' --sort -modifiedAt --limit 20
any query-subscribe SPACE CHAT --dataset chat_messages --sort -_ver.id --limit 50 --total \
  | jq 'select(.event=="changes") | .data[]'

What to subscribe to

• objects (cross-object) — one row per object holding computed property values. Creates land as added, property writes as updated, deletes as removed with reason: "deleted" — the canonical place to observe deletion across a space.
• editor_blocks — the block tree of one document; every frame ships the full post-apply block, whether it came from a block write or a bulk markdown import. See Editor.
• chat_messages — one message per added; a reaction toggle arrives as an updated event with a $set / $unset op under reactions.<emoji>.<accountId>. See Chat.

Every record ships its full stored form, including _ver (and _traces / _deletedAt when present). The projection field is accepted but not applied yet; strip system fields client-side if you want a leaner model. Hold one subscription per open view and size the window for the UI — the server streams straight from the indexed store and is indifferent to window size; the memory cost is the client's.