SapixDBSapixDB/Docs
Home
Community · Realtime

Realtime SSE Streams

Subscribe to live record writes on any agent using standard Server-Sent Events. No polling, no WebSocket handshake — open an HTTP connection and events arrive as they are written.

GET/v1/agents/:id/streamSubscribe to one named agent
GET/v1/streamSubscribe to all agents (optional ?agents= filter)
GET/v1/agents/:id/query/streamStream a query scan as SSE events
ℹ How it worksAll three write paths (JSON record, raw record, primary agent write) publish a record.written event onto a single in-process broadcast channel. SSE handlers subscribe to that channel and push matching events to connected clients. The broadcast call is synchronous and allocation-free — zero overhead when no clients are connected.

Per-agent stream

Opens a live stream scoped to one agent. Every record written to that agent is pushed as a record.written SSE event.

GET /v1/agents/:id/stream
GET /v1/agents/orders/stream
Authorization: Bearer <key>

Query parameters

ParamDescription
sinceHLC timestamp (ms). Backfills records written after this timestamp before going live. Capped at 1 000 most recent records.
filterOnly emit events whose JSON payload contains this top-level field name.
curl
# Live stream
curl -N http://localhost:7475/v1/agents/orders/stream

# Backfill from a known cursor, then live
curl -N "http://localhost:7475/v1/agents/orders/stream?since=1748304000000"

# Only events that have a risk_score field
curl -N "http://localhost:7475/v1/agents/transactions/stream?filter=risk_score"

Global stream

Subscribes to all agents on the instance. Pass an optional comma-separated agents list to narrow the stream.

GET /v1/stream
GET /v1/stream
Authorization: Bearer <key>

# Narrow to specific agents
GET /v1/stream?agents=orders,payments,inventory

Supports the same since and filter parameters as the per-agent stream.

Event format

All live-stream events use the record.written SSE event type:

SSE event
event: record.written
data: {
  "agent_id":     "orders",
  "event_type":   "record.written",
  "record_id":    "018f3c2a-4b1d-7e8f-a3c2-1d4e5f6a7b8c",
  "content_hash": "3a7bd3f1c2e9a4b5...",
  "timestamp_ms": 1748304000000,
  "payload":      { "order_id": "ord_123", "total": 4999 }
}

A heartbeat event is sent every 30 seconds to keep connections alive through proxies and load balancers:

heartbeat
event: heartbeat
data: { "event": "heartbeat", "subscribers": 4 }
FieldTypeDescription
agent_idstringThe agent that received the write
event_typestringAlways "record.written"
record_idstringUUID of the new record
content_hashstringSHA-256 content hash of the record payload
timestamp_msnumberHLC timestamp in milliseconds
payloadobject | nullDecoded JSON payload, if the record was written as JSON

Backfill and cursor resumption

Pass ?since=<timestamp_ms> to replay records written after a known HLC timestamp before receiving live events. SapixDB subscribes to the broadcast channel first, then replays historical records — so no event written between the two steps can be missed.

TypeScript — resume from last seen
let cursor = localStorage.getItem("stream_cursor") ?? "0";

const es = new EventSource(
  `/v1/agents/orders/stream?since=${cursor}`,
  { withCredentials: false }
);

es.addEventListener("record.written", (e) => {
  const event = JSON.parse(e.data);
  cursor = String(event.timestamp_ms);
  localStorage.setItem("stream_cursor", cursor);
  handle(event);
});
⚠ Backfill capBackfill is capped at 1 000 records per reconnect. If your client was offline long enough to miss more than 1 000 records, use GET /v1/agents/:id/strand/records?after=<cursor> to catch up in pages before opening the stream.

Streaming query scan

Stream the results of a filtered scan as SSE events instead of waiting for the entire response to buffer. Useful for large result sets or progressive rendering.

GET /v1/agents/:id/query/stream
GET /v1/agents/events/query/stream?type=scan&field=status&op=eq&value=pending&limit=500
Authorization: Bearer <key>
ParamDescription
typeQuery type — only scan is supported (default)
fieldFilter field name (optional)
opFilter operator: eq · ne · gt · lt · gte · lte · between · contains · starts_with · ends_with · like · fts · is_null · is_not_null
valueFilter value (parsed as JSON, otherwise treated as string)
upperUpper bound for the between operator
limitMax records to stream (default 100)

Each record is emitted as a plain data: event carrying the full RecordView JSON. A done event signals the end of the stream:

event format
data: {"record_id":"...","content_hash":"...","payload":{...},"timestamp_ms":...}

event: done
data: {}
TypeScript
const es = new EventSource(
  "/v1/agents/events/query/stream?type=scan&field=status&op=eq&value=pending"
);

es.onmessage = (e) => {
  const record = JSON.parse(e.data);
  renderRow(record);
};

es.addEventListener("done", () => es.close());
ℹ Single predicate onlyThe query stream accepts a single leaf filter predicate. For compound filters (AND / OR / nested conditions) use POST /v1/agents/:id/query instead.

TypeScript SDK

The SapixDB TypeScript SDK wraps the raw EventSource API:

TypeScript SDK
import { createAgentStream, createGlobalStream } from "@sapixdb/sdk";

// Subscribe to one agent
const es = createAgentStream("orders", (event) => {
  console.log("new record:", event.record_id, event.payload);
});

// Backfill from a cursor, then go live
const es = createAgentStream("orders", handler, { since: lastSeenHlc });

// Watch multiple agents
const es = createGlobalStream(handler, { agents: ["orders", "payments"] });

// Filter to events that contain a specific field
const es = createGlobalStream(handler, { filter: "risk_score" });

// Always close when done
es.close();

Limitations

LimitationDetail
Slow consumers drop eventsThe broadcast channel holds 1 024 events. A client that cannot keep up will miss events rather than blocking writes. Use ?since= to recover any gap on reconnect.
Backfill capped at 1 000 recordsSee the backfill section above.
payload is null for raw (msgpack) writesOnly JSON writes include a decoded payload. Raw binary records set payload to null.
No per-field filtering on the live streamThe ?filter= param matches event payloads that contain the named field — it does not support value comparisons. Use the query stream for predicate filtering.
Primary agent writes only for /v1/streamThe global stream and per-agent stream broadcast from the primary write path. Writes via /v1/agents/:id/records/:agent are also broadcast.
← Connector HubMulti-Region Replication →