Publications

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
}

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
}

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

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

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