@rotorsoft/act-sse

@rotorsoft/act-sse

Incremental state broadcast over SSE for @rotorsoft/act event-sourced apps. Zero dependencies.

Instead of sending full aggregate state after each action, act-sse forwards the domain patches that event handlers already compute — sending only what changed as version-keyed partials.

Installation

npm install @rotorsoft/act-sse
# or
pnpm add @rotorsoft/act-sse

Requirements: Node.js >= 22.18.0

Architecture

  app.do() → Snapshot[] (each carries its event's domain patch)
      │
      ▼
  deriveState(snap)              ← app-specific (overlay presence, deadlines, etc.)
  state._v = snap.event.version  ← event store version is the single source of truth
      │
      ▼
  broadcast.publish(streamId, state, patches)
      │
      └── version-key each patch → { "5": { count: 3 }, "6": { name: "updated" } }
      └── push to all SSE subscribers
      │
      ▼
  Client: applyPatchMessage(msg, cached)
      │
      ├── contiguous → deep merge patches in version order
      ├── stale      → skip (client already ahead)
      └── behind     → invalidate + refetch (client missed versions)

Wire Format

// Version-keyed domain patches
// Keys = state version after that patch is applied
// Values = domain patch (deep partial of state)
{
  "5": { territories: { brazil: { armies: 3 } } },
  "6": { currentPlayerIndex: 2, phase: "reinforce" }
}

Multi-event commits produce multiple version-keyed entries. Version gaps trigger full state refetch on the client.

Server Usage

import { BroadcastChannel } from "@rotorsoft/act-sse";

const broadcast = new BroadcastChannel<MyAppState>();

// After every app.do():
const snaps = await app.do(action, target, payload);
const snap = snaps.at(-1)!;
const patches = snaps.map(s => s.patch).filter(Boolean);
const state = deriveState(snap);
broadcast.publish(streamId, state, patches);

// For non-event state changes (e.g. presence overlay):
broadcast.publishOverlay(streamId, { players: { pid: { connected: true } } });

// SSE subscription (tRPC example):
onStateChange: publicProcedure
  .input(z.object({ streamId: z.string() }))
  .subscription(async function* ({ input, signal }) {
    let resolve: (() => void) | null = null;
    let pending: PatchMessage | null = null;

    const cleanup = broadcast.subscribe(input.streamId, (msg) => {
      pending = msg;
      if (resolve) { resolve(); resolve = null; }
    });

    try {
      while (!signal?.aborted) {
        if (!pending) {
          await new Promise<void>((r) => {
            resolve = r;
            signal?.addEventListener("abort", () => r(), { once: true });
          });
        }
        if (signal?.aborted) break;
        if (pending) { const msg = pending; pending = null; yield msg; }
      }
    } finally {
      cleanup();
    }
  }),

Client Usage

import { applyPatchMessage } from "@rotorsoft/act-sse";

// In your SSE onData handler (React Query example):
onData: (msg) => {
  const cached = utils.getState.getData({ streamId });
  const result = applyPatchMessage(msg, cached);

  if (result.ok) {
    utils.getState.setData({ streamId }, result.state);
  } else if (result.reason === "behind") {
    // Client missed versions — trigger full refetch
    utils.getState.invalidate({ streamId });
  }
  // "stale" → no-op (client already has newer state from mutation response)
}

API

BroadcastChannel<S>

Server-side broadcast manager with per-stream subscriber channels and LRU state cache.

Method	Description
publish(streamId, state, patches?)	Cache state, version-key patches, push to subscribers
publishOverlay(streamId, patch)	Same-version update (e.g. presence change)
subscribe(streamId, cb)	Register subscriber, returns cleanup function
getState(streamId)	Get cached state (for reconnects)
getSubscriberCount(streamId)	Number of active subscribers
cache	Direct access to StateCache instance

applyPatchMessage(msg, cached)

Client-side patch applicator. Returns { ok: true, state } or { ok: false, reason }.

Reasons: "stale" (skip), "behind" (resync).

patch(original, patches)

Browser-safe deep merge utility. Deep merges plain objects, replaces arrays and other non-mergeable types (Date, Map, Set, RegExp, TypedArrays). Deletes keys set to null or undefined.

StateCache<S>

Generic LRU cache. Methods: get, set, delete, has, size, entries.

Types

type BroadcastState = Record<string, unknown> & { _v: number };
type PatchMessage<S extends BroadcastState> = Record<number, DeepPartial<S>>;
type Subscriber<S extends BroadcastState> = (msg: PatchMessage<S>) => void;

Version Contract

The _v field must be set from snap.event.version — the event store's monotonic stream version. This is the single source of truth for ordering. No separate version counters.

Bandwidth Savings

Typical savings for a game/collaborative app with ~5KB state:

Action	Full (bytes)	Patch (bytes)	Savings
Single field change	~5,000	~150	97%
Multi-field update	~5,000	~400	92%
Presence toggle	~5,000	~80	98%

Related

• @rotorsoft/act - Core framework
• @rotorsoft/act-pg - PostgreSQL adapter
• Documentation
• Examples

License

MIT