Live event streams

The gateway streams events as they happen: chain activity, wallet transfers and approvals, issuer actions. Apps can react without polling. There are two transports over one event store: Server-Sent Events for browsers and simple clients, and WebSocket for several subscriptions over one connection. Both resume from the last event they delivered.

Events

Every event carries the same JSON payload as gateway webhooks:

{"id": "evt_wallet_3f2a91c0_412", "seq": 6593, "type": "wallet.transfer.approved", "network": "regtest",
 "source": "wallet", "subject": "wal_1c…", "height": 0, "created": 1792300000,
 "data": {"subject": "wal_1c…", "actor": "key_…", "action": "transfer.approved", "at": "…", "detail": {…}}}

Source	Types	Who receives them
chain	`block.connected`, `block.reorged`, `address.activity`, `asset.issued` / `minted` / `burned` / `frozen` / `unfrozen` / `clawback`, `event.retracted`	Everyone
wallet	`wallet.created`, `wallet.keys.registered`, `wallet.transfer.created` / `approved` / `rejected` / `signed` / `broadcast` / `confirmed` / `denied` / `expired` / `failed` / `cancelled`, `wallet.policy.`, `wallet.frozen`, `wallet.allowlist.`, `wallet.coins.*`	The wallet's owner, approvers and co-signer principals
issuer	`issuer.asset.registered`, `issuer.claim.`, `issuer.holder.registered`, `issuer.assignment.created`, `issuer.distribution.created`, `issuer.action.`, `issuer.dividend.*`, `issuer.attestation.published`	The issuer's officers

seq is the stream position and the SSE id. It increases strictly, and resuming "after N" never skips an event, because the gateway serialises inserts so sequence numbers become visible in commit order.
Delivery is at least once. Deduplicate on id.
How wallet and issuer events get here: each service writes its audit row in the same database transaction as the state change (a transactional outbox), and a relay forwards it to the gateway. An event is therefore emitted if and only if the change committed.
Reorgs: every event of an orphaned block is followed by an event.retracted that names it (data.retracted_event_id, data.retracted_kind). The retraction keeps the original's addresses and assets, so a filtered stream receives it too. A type filter for address.activity also delivers retractions of address.activity.

Filters

Filter	Example	Applies to
`types`	`wallet.*,block.connected,address.activity`	All events. Accepts exact types, ``, and prefixes such as `wallet.transfer.`
`addresses`	`rarrk1…,rarrk1…`	Chain events (block-level events always pass)
`assets`	`<asset hex>`	Chain events
`subjects`	`wal_…,iss_…`	Wallet and issuer events

Authentication

Server-side clients send X-API-Key: arrk_…, as with every other gateway route.
Browsers can't set headers on EventSource or WebSocket. Exchange the key server-side for a short-lived, stream-only token and put that in the URL:
```
POST /gateway/v1/stream/tokens  {"ttl_secs": 3600}      →  {"token": "st_…", "expires_at": "…", "scope": "streams:read"}
```
Tokens last 60 s to 24 h (default 1 h). They can only open streams, and revoking the API key kills them at once. A long-lived key never appears in a URL or access log.
Limits: each key may hold 20 open streams, and each WebSocket 10 subscriptions.

Server-Sent Events

const es = new EventSource(`/gateway/v1/stream?token=${token}&types=wallet.*,block.connected`);
es.addEventListener("wallet.transfer.approved", (e) => handle(JSON.parse(e.data)));
es.addEventListener("stream.reset", (e) => resyncFromRest(JSON.parse(e.data)));

Opening: the stream begins with stream.open ({"cursor": N}) and a retry: 3000 hint. A : keepalive comment follows every 15 s.
Resume: EventSource sends Last-Event-ID automatically when it reconnects, and the stream continues strictly after that event. Other clients can pass ?after=<seq>. after=0 replays all retained history. With no cursor the stream starts live.
Expired cursors: history is kept for ARREKTA_GATEWAY_EVENT_RETENTION_DAYS (default 7). A cursor older than that gets stream.reset ({"reason": "cursor_expired", "oldest_available": …}) instead of a silent gap. Rebuild state over REST, then carry on from the stream.

WebSocket

GET /gateway/v1/ws (with an X-API-Key header, or ?token=st_…). JSON messages:

→ {"op": "subscribe", "id": "s1", "types": ["wallet.transfer.*"], "subjects": ["wal_…"], "after": 6500}
← {"op": "subscribed", "id": "s1", "cursor": 6500}
← {"op": "event", "sub": "s1", "seq": 6593, "event": { … }}
← {"op": "reset", "sub": "s1", "detail": {"reason": "cursor_expired", …}}
→ {"op": "unsubscribe", "id": "s1"}        ← {"op": "unsubscribed", "id": "s1", "existed": true}
→ {"op": "ping"}                           ← {"op": "pong"}

Cursors: each subscription keeps its own cursor. Store the last seq per subscription and pass it as after when you reconnect.
Liveness: the server pings every 20 s and closes connections that have been silent for 60 s (half-open detection).
Message size: client messages are limited to 64 KB.

Backpressure

Streams pull from the event store by cursor rather than buffering pushed messages. A slow consumer falls behind (memory stays bounded) and catches up from where it is.

Webhooks

Webhooks accept the same patterns (wallet.*, issuer.action.*, exact types). Wallet and issuer events reach only webhooks owned by a key in the event's audience. They are signed like chain events (ARRK-Signature: v1,ml-dsa-65=…).

Operations

Internal ingest: POST /internal/v1/events takes events from the wallet and issuer outbox relays and requires the service token. The relays are enabled with ARREKTA_WALLET_GATEWAY_URL and ARREKTA_ISSUER_GATEWAY_URL (set in compose).
Multiple gateway instances: streams wake on an in-process notification and also poll the store every 2 s, so several gateway instances behind a load balancer all stay correct.
Check: core/.venv/bin/python all_svc/stream_check.py (17 end-to-end checks).