Skip to main content

Eventing & SSE

Server-sent events are the live-sync seam between the single-writer store and any number of readers (Flutter LiveSyncService, browser tabs, curl).

Types​

EventType: charge.created, charge.updated, task.created, notification.updated, sync.required — frozen in contract v1. ResourceType: charge, task, notification. ChangeType: created, updated. Every commit appends DraftEvent → StoredEvent{ id, type, entityId, resourceType, changeType, bookingId?, entityVersion?, createdAt, streamCursor } to the ring.

Ring + cursors​

  • Capacity EVENT_RING_CAPACITY=100 — oldest dropped when full, never blocking commit.
  • streamCursor is an opaque monotonic id (event_id() using IdGenerator); get_sync_snapshot_meta_200.json shows shape.
  • streamEpoch changes only on reset — signals full refetch.

SSE endpoint​

GET /api/v1/events
Accept: text/event-stream
Last-Event-ID: 01H...   # optional; replays from that cursor
Cache-Control: no-cache

Response frames (see docs/contracts/examples/sse_frames.json):

id: 01HROTZ3Q...
event: charge.updated
data: {"chargeId":"CHG-001","status":"contested","bookingId":"BKG-001","...": "..."}

: heartbeat

event: sync.required
data: {"reason":"ring trimmed; cursor not retained"}
  • EventBroker in app/services/event_bus.py holds a per-client asyncio.Queue (capacity sse_queue_capacity=64, default 64). On overflow oldest is dropped per-client.
  • publisher(events) is set by store.set_publisher(broker.publish) in lifespan; called after os.replace so clients never see uncommitted events.
  • GET /events checks Last-Event-ID against the ring; if found, replays from next event; if not found (trimmed/unknown) sends sync.required so client takes GET /sync-snapshot.
  • Heartbeat comment ": \n\n" every sse_heartbeat_seconds=15 to keep proxies from closing.
  • Barrier: /events is in _SELF_MANAGED_PATHS — it bypasses the OperationBarrier.shared() so a long-lived stream doesn't block reset; reset closes all broker queues via await broker.close_all().

Flutter consumption​

LiveSyncService (lib/core/sync/live_sync_service.dart) opens KxApi.subscribeEvents(cursor) using dio with ResponseType.stream; SseEntityCommitter (sse_entity_committer.dart) applies charge.updated to Drift via LocalStore.applyRemoteChanges, then advances SnapshotCoordinator cursor. SnapshotCoordinator decides when sync.required forces a full GET /sync-snapshot. Leader election (SyncLeadership — sync_leadership.dart + factory_web/stub) ensures one tab drains the stream when multiple browser tabs are open.

Chaos & failure​

ChaosController (app/core/chaos.py) can inject per-path latency and probabilistic 429 via KX_CHAOS_LATENCY_MS / KX_CHAOS_ERROR_RATE toggles (also surfaced to Flutter --dart-define). SSE is injectable — LiveSyncService must retry with backoff (RetryPolicy).

Testing​

  • tests/api/test_attachments_events_dev.py — broker queue capacity, heartbeat, replay via Last-Event-ID, ring trim detection.
  • Golden sse_frames.json is the bytes-level oracle; contract tests assert header text/event-stream + cache-control.

What SSE is not​

This is not a durable log — 100 events, no persistence beyond state.json ring. Not a message queue — not acknowledged, no delivery guarantee. It is a notification that something changed; authoritative state is always GET /sync-snapshot or GET /hub + charge endpoints.