SapixDBSapixDB/Docs
Home
Community · Change Feed

Publications

Subscribe to agent writes as a named, durable change feed. Each consumer gets its own slot with a server-tracked cursor — reconnects resume automatically from where the consumer left off.

📡
Named publications
Define a publication once with a name and a set of agents. Multiple consumers subscribe independently without re-specifying the filter.
🔖
Durable slot cursors
Each consumer registers a slot. SapixDB tracks the cursor server-side and bacfkfills automatically on reconnect — no client-side cursor management.
🪦
Tombstone visibility
Every event carries a flags field. flags=2 means TOMBSTONE (logical delete), giving consumers full DELETE visibility — not just inserts.
⬡
Multiple consumers
Any number of slots on the same publication. Each cursor advances independently — a slow consumer does not block a fast one.

Publications vs. raw SSE stream

Raw SSE streamPublications
NamedNoYes
Agent scopingAd-hoc (?agents=)Defined on the publication
Cursor trackingClient-sideServer-side per slot
Resume on reconnectManual ?since=Automatic from slot cursor
Multiple independent consumersNoYes — one slot per consumer
Tombstone visibilityflags fieldflags field

Quick start

Three steps: create a publication, create a slot, open the stream.

1 — create publication
POST /v1/publications
Authorization: Bearer <key>
Content-Type: application/json

{
  "name": "order-events",
  "agents": ["orders", "payments"]
}

// Response 201
{
  "id": "a1b2c3d4",
  "name": "order-events",
  "agents": ["orders", "payments"],
  "enabled": true,
  "created_at_ms": 1754478000000,
  "event_count": 0,
  "last_event_ms": null
}
2 — create slot
POST /v1/publications/a1b2c3d4/slots
Authorization: Bearer <key>
Content-Type: application/json

{
  "name": "analytics-consumer",
  "cursor_hlc": 0
}

// Response 201
{
  "id": "slot_xyz",
  "publication_id": "a1b2c3d4",
  "name": "analytics-consumer",
  "cursor_hlc": 0,
  "created_at_ms": 1754478060000,
  "last_active_ms": null
}
3 — open stream
GET /v1/publications/a1b2c3d4/stream?slot=slot_xyz
Authorization: Bearer <key>

// SSE events arrive as records are written to orders or payments:
event: record.written
data: {
  "publication_id": "a1b2c3d4",
  "slot_id":        "slot_xyz",
  "agent_id":       "orders",
  "record_id":      "018f3c2a-...",
  "content_hash":   "3a7bd3f1...",
  "flags":          0,
  "cursor":         1754480042000,
  "event_type":     "record.written",
  "payload":        { "order_id": "ord_123", "total": 4999 }
}

Event format

FieldTypeDescription
publication_idstringPublication this event belongs to
slot_idstringThe consuming slot whose cursor was just advanced
agent_idstringAgent that received the write
record_idstringUUID of the written record
content_hashstringSHA-256 content hash of the record payload
flagsnumber0 = normal write · 2 = TOMBSTONE (logical delete)
cursornumberHLC timestamp (ms) — new cursor value after this event
payloadobject | nullDecoded JSON payload; null for raw msgpack writes
ℹ TOMBSTONE eventsWhen flags === 2, the record is a logical deletion marker. The original record is not removed from the immutable strand — a TOMBSTONE is appended after it. Consumers should treat these as delete events and remove the item from their downstream store.

TypeScript — durable consumer

Create the publication and slot once. On every subsequent deploy or restart, just open the stream — SapixDB resumes from the last confirmed cursor automatically.

TypeScript
const BASE = "https://your-instance.sapixdb.com";
const KEY  = process.env.SAPIX_API_KEY!;

async function ensurePublication() {
  // Idempotent: check if it exists first
  const list = await fetch(`${BASE}/v1/publications`, {
    headers: { Authorization: `Bearer ${KEY}` },
  }).then(r => r.json());

  const existing = list.publications.find((p: { name: string }) => p.name === "order-events");
  if (existing) return existing;

  return fetch(`${BASE}/v1/publications`, {
    method: "POST",
    headers: { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify({ name: "order-events", agents: ["orders", "payments"] }),
  }).then(r => r.json());
}

async function ensureSlot(pubId: string, slotName: string) {
  const list = await fetch(`${BASE}/v1/publications/${pubId}/slots`, {
    headers: { Authorization: `Bearer ${KEY}` },
  }).then(r => r.json());

  const existing = list.slots.find((s: { name: string }) => s.name === slotName);
  if (existing) return existing;

  return fetch(`${BASE}/v1/publications/${pubId}/slots`, {
    method: "POST",
    headers: { Authorization: `Bearer ${KEY}`, "Content-Type": "application/json" },
    body: JSON.stringify({ name: slotName, cursor_hlc: 0 }),
  }).then(r => r.json());
}

function openStream(pubId: string, slotId: string) {
  const es = new EventSource(
    `${BASE}/v1/publications/${pubId}/stream?slot=${slotId}`
    // Pass auth via cookie or a proxy that injects the Authorization header
  );

  es.addEventListener("record.written", (e) => {
    const event = JSON.parse(e.data);

    if (event.flags === 2) {
      console.log("DELETE", event.agent_id, event.content_hash);
      handleDelete(event);
    } else {
      console.log("WRITE", event.agent_id, event.payload);
      handleWrite(event);
    }
    // cursor advances server-side automatically — no ack needed
  });

  es.addEventListener("error", () => {
    es.close();
    setTimeout(() => openStream(pubId, slotId), 2_000);
    // Reconnect resumes from the last cursor — no data lost
  });
}

const pub  = await ensurePublication();
const slot = await ensureSlot(pub.id, "analytics-consumer");
openStream(pub.id, slot.id);

Full API reference

Publications

MethodPathDescription
POST/v1/publicationsCreate a publication
GET/v1/publicationsList all publications
GET/v1/publications/:idGet one publication
PATCH/v1/publications/:idUpdate name, agents, or enabled state
DELETE/v1/publications/:idDelete publication and all its slots
GET/v1/publications/:id/stream?slot=<id>SSE stream for a subscriber slot

Subscriber slots

MethodPathDescription
POST/v1/publications/:id/slotsCreate a slot (cursor_hlc defaults to 0)
GET/v1/publications/:id/slotsList all slots for this publication
GET/v1/publications/:id/slots/:slot_idGet one slot
DELETE/v1/publications/:id/slots/:slot_idDelete a slot
POST/v1/publications/:id/slots/:slot_id/ackManually advance cursor to { cursor: <hlc> }

PATCH fields

PATCH /v1/publications/:id
{
  "name":    "new-name",         // optional
  "agents":  ["orders"],         // optional — pass null to switch to all-agents
  "enabled": false               // optional — false disables the stream endpoint
}

How cursor tracking works

// 1. client opens stream with slot=slot_xyz
// 2. server subscribes to broadcast bus (no events missed)
// 3. server bacfkfills strand records since slot.cursor_hlc
// 4. cursor advances to last backfilled event
// 5. live events stream in; cursor advances after each delivery
// 6. client disconnects — cursor persisted in graph meta
// 7. client reconnects — go to step 2, backfill from cursor
⚠ Backfill capBackfill is capped at 1 000 records per agent per reconnect. If a slot falls more than 1 000 records behind on any agent, page-catch-up first using GET /v1/agents/:id/strand/records?after=<cursor>, then open the stream.

Limitations

At-most-once delivery for live events
The broadcast bus holds 1 024 events. A slow consumer that cannot keep up will miss live events. Cursor only advances for delivered events, so the gap is recovered on reconnect from the strand.
Cursor advances on delivery, not ACK
The SSE stream auto-advances the cursor as events are sent. For explicit at-least-once semantics, poll the strand directly and call POST .../ack manually.
No per-field filtering
All events from covered agents are delivered. Filtering is the consumer's responsibility.
payload is null for raw msgpack writes
Only JSON-written records include a decoded payload. Raw binary records set payload to null; use the content_hash to fetch the full record.
← Realtime StreamConnector Hub →