SSE / WebSocket

causet-realtime fans committed ledger patches and projection writes to connected clients over WebSocket or Server-Sent Events (SSE).

It does not execute intents or own entity state. It is a read path for live updates — complementary to named queries (poll/HTTP) and entity state (one-shot fetch).

For durable export to your Kafka cluster or HTTPS endpoints (jobs, retries, Delivery Log), use Destinations. Live WebSocket/SSE is for connected clients; destinations are for systems that must not miss events.

⚠️

Subscribe only. The Realtime service never accepts intents. Submit intents with submitIntent() via the Causet management API (CAUSET_API_URL, local default http://localhost:8085). Inbound webhook handlers verify the payload, then call submitIntent() — the webhook is not the transport. Do not use submitIntentStream() in webhook handlers or short-lived request paths; it holds the connection open until execution finishes.

Where it fits

Causet separates writes from live reads:

Path	Service (local)	Port	What it does
Write (SDK default)	Causet management API	8085	Intent submission, token exchange, entity state
Write (advanced)	Runtime API	8080	Direct runtime HTTP for local debugging
Read (query)	Query service	8082	Named projection queries
Read (live)	Realtime service	8081	WebSocket / SSE fan-out

submitIntent() ──▶ Causet management API (:8085)
                         │
                         ├──▶ Runtime API (:8080) — commits events, updates snapshots
                         │         │
                         │         ├──▶ Kafka: causet.patches.v1
                         │         │         │
                         │         │         ▼
                         │         │    Realtime service (:8081) ──▶ WebSocket / SSE clients
                         │         │
                         │         ├──▶ Projection worker (:8083) ──▶ projection tables
                         │         │                                      │
                         │         │                                      ▼
                         │         └──▶ Query service (:8082) ◀── HTTP query API

Implementation note (local Docker): The Compose stack uses internal container hostnames. Application code should use the public service names and localhost ports above.

Kafka topics consumed by causet-realtime:

causet.patches.v1 — ledger patches (state mutations, domain events)
causet.projection-writes.v1 — materialized projection row updates

Base URLs

The Realtime HTTP host is separate from the management API host. SDKs derive it automatically from apiUrl:

Environment	API URL	Realtime HTTP	WebSocket
Sandbox	`https://sandbox.api.causet.cloud`	`https://sandbox.realtime.causet.cloud`	`wss://sandbox.realtime.causet.cloud/ws`
Prod	`https://api.causet.cloud`	`https://realtime.causet.cloud`	`wss://realtime.causet.cloud/ws`
Local	`http://localhost:8085`	`http://localhost:8081`	`ws://localhost:8081/ws`

Local stack: docker compose up -d causet-realtime from the Causet repo.

Authentication

Clients authenticate with a short-lived JWT minted by the Causet management API:

curl -s -X POST https://sandbox.api.causet.cloud/v1/token \
  -H "Authorization: Bearer ck_live_..." | jq -r .token

The API key must include the WEBSOCKET_READ scope (or legacy STREAM_SUBSCRIBE).

SDKs call /v1/token automatically in init() and refresh the JWT before expiry. Pass the token to causet-realtime via:

WebSocket: Authorization: Bearer <jwt> header or ?token=<jwt> query param
SSE: Authorization: Bearer <jwt> header or ?token=<jwt> query param

platform_id and application_id are derived from the JWT — do not send them in subscribe messages.

Subscription modes

Subscriptions are scoped by stream, fork, and optionally entity:

Mode	`stream_id`	`fork_id`	Receives
Stream + fork	`sku_stream`	`sandbox`	All entities on that stream/fork
Stream + fork + entity	`sku_stream:sku-1`	`sandbox`	Only `sku-1` on that fork

Use from_cursor to replay missed events from the runtime ledger, then continue with live events. Cursor 0 starts from the beginning; pass last cursor + 1 to resume.

WebSocket

Endpoint: {realtimeBase}/ws (default path /ws, configurable via WS_PATH)

Connection flow

Client opens WebSocket with JWT
Client sends hello within 10s (SDK sends this automatically on connectStream)
Server sends welcome
Server streams replay events (if from_cursor set), then live Kafka events
Client may send sub, unsub, or ping

Client → server: hello

Sent automatically by SDKs on connect:

{
  "type": "hello",
  "v": 1,
  "stream_id": "sku_stream",
  "fork_id": "sandbox",
  "subs": [
    { "channel": "ledger", "from_cursor": 0 },
    { "channel": "state", "from_cursor": 0 }
  ]
}

Field	Description
`stream_id`	Stream name, or `streamType:entityId` for a single entity
`fork_id`	Fork to listen on (default `main`)
`subs`	Optional channel subscriptions (`ledger`, `state`) with optional `from_cursor`

Server → client: welcome

{
  "type": "welcome",
  "v": 1,
  "conn_id": "550e8400-e29b-41d4-a716-446655440000",
  "server_ts": 1709068800000,
  "shard": 0
}

Server → client: ledger patch

Events are forwarded verbatim from Kafka — no type: "event" wrapper. Ledger events have cursor, stream_id, and platform_id:

{
  "cursor": 42,
  "platform_id": "org_xxx",
  "application_id": "app_xxx",
  "stream_id": "orders",
  "entity_id": "order-123",
  "fork_id": "sandbox",
  "intent_id": "intent_xxx",
  "event_type": "ORDER_PLACED",
  "patch": [
    { "op": "replace", "path": "/status", "value": "placed" },
    { "op": "add", "path": "/items/-", "value": { "sku": "prod-1", "qty": 2 } }
  ],
  "emits": [
    { "event_type": "ORDER_PLACED", "payload": { "amount": 99.99 } }
  ],
  "metadata": {
    "causation_id": "intent_xxx",
    "correlation_id": "uuid-...",
    "ruleset_version": 1,
    "deployed_version": 1,
    "ts": "2025-02-27T12:00:00Z"
  }
}

patch: JSON Patch RFC 6902 operations (replace, add, remove)
emits: Domain events emitted by rules
event_type: Business event (e.g. ORDER_PLACED, SET, EMIT)

Server → client: projection write

{
  "stream_id": "orders",
  "projection_name": "inventory",
  "entity_id": "order-123",
  "platform_id": "org_xxx",
  "application_id": "app_xxx",
  "fork_id": "main",
  "cursor": 42,
  "event_type": "projection:inventory",
  "trigger_event_type": "ORDER_PLACED",
  "row": {
    "product_id": "prod-1",
    "quantity": 5,
    "updated_at": "2025-02-27T12:00:00Z"
  },
  "ts": "2025-02-27T12:00:00Z"
}

row: The materialized projection row
trigger_event_type: The ledger event that caused this update
event_type: projection:{projectionName} for filtering

Server → client: error

{
  "type": "error",
  "code": "FORBIDDEN",
  "message": "stream access denied"
}

Code	When
`UNAUTHORIZED`	No token, invalid JWT, or JWT expired
`FORBIDDEN`	Wrong project/env, stream or channel access denied
`BAD_REQUEST`	Invalid hello, invalid `stream_id` format, too many subs
`NOT_FOUND`	Stream not found (subscribe with `from_cursor`)
`RATE_LIMITED`	Too many inbound messages
`INTERNAL`	Internal error, subscription failed

Distinguishing message types

Protocol messages have top-level "type" in welcome, error, redirect, pong
Ledger events have cursor, stream_id, platform_id — no protocol type
Projection events have projection_name, row, and event_type starting with projection:

SSE (Server-Sent Events)

Endpoint:

GET {realtimeBase}/v1/platforms/{platformId}/applications/{applicationId}/streams/{streamId}/events
    ?fork_id={fork}&from_cursor={cursor}&token={jwt}

Param	Description
`streamId`	Stream name or `streamType:entityId`
`fork_id`	Fork to subscribe to (default `main`)
`from_cursor`	Replay from this cursor; omit or `0` for from start
`token`	JWT (alternative to `Authorization` header)

Headers: Accept: text/event-stream, Authorization: Bearer <jwt>

Wire format

Each event:

id: 42
event: STOCK_ADJUSTED
data: {"cursor":42,"stream_id":"sku_stream","entity_id":"sku-1","fork_id":"sandbox","event_type":"STOCK_ADJUSTED","patch":[{"op":"replace","path":"/quantity","value":95}]}

id: Cursor for ordering and resume
event: Event type string (e.g. ORDER_PLACED, projection:inventory)
data: Raw JSON payload — same shape as WebSocket ledger/projection events

Heartbeat: the server sends : ping comments every ~15s to keep the connection alive.

When to use which transport

Transport	Best for
`websocket`	Duplex subscriptions, `ledger` + `state` channel replay, lowest latency
`sse`	One-way fanout, browser `EventSource`, mobile, HTTP-only environments

SDK usage

Every SDK exposes connectStream() (or connect_stream()) with the same subscription model. Initialize the client first so the JWT is cached.

TypeScript / JavaScript

import { CausetClient } from '@causet/sdk';
 
const client = new CausetClient({
  apiUrl: 'https://sandbox.api.causet.cloud',
  platformSlug: 'my-platform',
  appSlug: 'my-app',
  forkId: 'sandbox',
  apiKey: process.env.CAUSET_API_KEY,
});
 
await client.init();
 
// WebSocket — stream + fork (all entities)
const connId = await client.connectStream('sku_stream', {
  fromCursor: 0,
  channels: [{ channel: 'ledger' }, { channel: 'state' }],
});
 
client.on('stream_event', (ev) => {
  console.log(ev.event_type, ev.entity_id, ev.cursor);
});
 
client.on('stream_connected', ({ connId, transport }) => {
  console.log(`Connected via ${transport}: ${connId}`);
});
 
// SSE — stream + fork + entity, resume from cursor 100
await client.connectStream('sku_stream:sku-1', {
  transport: 'sse',
  fromCursor: 100,
});
 
client.disconnectStream();

Next.js apps use the same client via @causet/sdk-next — see TypeScript / JavaScript SDK.

Python

from causet_sdk import CausetClient
 
client = CausetClient(
    api_url="https://sandbox.api.causet.cloud",
    platform_slug="my-platform",
    app_slug="my-app",
    fork_id="sandbox",
    api_key="ck_live_...",
)
 
await client.init()
 
conn_id = await client.connect_stream(
    "sku_stream",
    from_cursor=0,
    channels=[{"channel": "ledger"}, {"channel": "state"}],
)
 
client.on("stream_event", lambda ev: print(ev.get("event_type"), ev.get("cursor")))
 
await client.connect_stream(
    "sku_stream:sku-1",
    transport="sse",
    from_cursor=100,
)
 
client.disconnect_stream()

Go

client, _ := causet.NewClient(causet.Config{
    APIURL:       "https://sandbox.api.causet.cloud",
    PlatformSlug: "my-platform",
    AppSlug:      "my-app",
    ForkID:       "sandbox",
    APIKey:       os.Getenv("CAUSET_API_KEY"),
})
client.Init(ctx)
 
connID, _ := client.ConnectStream(ctx, "sku_stream", causet.StreamConnectOptions{
    FromCursor: 0,
    Channels:   []causet.StreamChannel{{Channel: "ledger"}, {Channel: "state"}},
}, func(ev map[string]any) {
    fmt.Println(ev["event_type"], ev["entity_id"], ev["cursor"])
})
 
connID, _ = client.ConnectStream(ctx, "sku_stream:sku-1", causet.StreamConnectOptions{
    Transport:  causet.StreamTransportSSE,
    FromCursor: 100,
}, onEvent)

Java

CausetClient client = new CausetClient(
    "https://sandbox.api.causet.cloud",
    "my-platform", "my-app", "sandbox", "ck_live_...");
client.init();
 
client.connectStream("sku_stream", null, event -> {
    System.out.println(event.get("event_type") + " " + event.get("entity_id"));
}).join();
 
CausetClient.StreamConnectOptions opts = new CausetClient.StreamConnectOptions();
opts.transport = StreamTransport.SSE;
opts.fromCursor = 100;
client.connectStream("sku_stream:sku-1", opts, event -> {
    System.out.println(event.get("cursor"));
}).join();
 
client.disconnectStream();

PHP / Laravel

connectStream() is SSE-only (Laravel/Guzzle is synchronous — no wrapped WebSocket client). It blocks the calling process for the connection’s lifetime; run it from a queue job or artisan command, not a web request handler:

use Causet\Laravel\Facades\Causet;
 
Causet::init();
 
Causet::on('stream_event', function (array $ev): void {
    logger()->info($ev['event_type'] ?? 'event', $ev);
});
 
Causet::connectStream('sku_stream:sku-1', function (array $ev): void {
    logger()->info($ev['event_type'] ?? 'event', $ev);
}, forkId: 'sandbox', fromCursor: 100);

Stop the loop from a signal handler in long-running commands:

pcntl_async_signals(true);
pcntl_signal(SIGTERM, fn () => Causet::disconnectStream());
 
Causet::connectStream('sku_stream', $onEvent);

For intent submission, use submitIntent() from controllers, jobs, and webhook handlers. Intent-progress SSE (submitIntentStream) hits the management API stream endpoint — not the Realtime service — and is only for long-running server contexts where you can hold the connection open (not webhooks or typical SSR/page renders).

SDK event handlers

Event	When
`stream_connected`	WebSocket or SSE connection established
`stream_event`	Ledger patch or projection write received
`patch_op`	Individual JSON Patch operation from a ledger event

Call disconnectStream() / disconnect_stream() when done.

Health and operations

Endpoint	Purpose
`GET /healthz`	Liveness
`GET /readyz`	Readiness (checks Kafka)
`GET /metrics`	Prometheus metrics

Destinations — durable Kafka / webhook export
SDK Overview — package matrix and shared client shape
TypeScript SDK — connectStream, submitIntentStream, low-level transports
Entity State — write-side aggregates (what patches mutate)
Projections — read models (what projection writes materialize)
Debugging Timeline — how rules ran for a committed intent

Inbound webhooks Installation