Real-time Streams

Intent submit ──▶ causet-runtime-service
                         │
                         ├──▶ Kafka: causet.patches.v1
                         │         │
                         │         ▼
                         │    causet-realtime ──▶ WebSocket / SSE clients
                         │
                         ├──▶ causet-projection-worker ──▶ projection tables
                         │                                      │
                         │                                      ▼
                         └──▶ causet-query-service ◀── HTTP query API

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

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

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

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

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

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

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

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

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

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

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)

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

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

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