Realtime Engine & Event Model

Realtime Engine & Event Model

The RealtimeEngine is the headless, session-scoped entry point of the realtime-core package: it owns the signal connection, routes inbound events to subscribers, intercepts media signaling events for the SFU media layer, and exposes the chat/call API surface. The event model defines how wire-level SignalEvents (bigint-based) are normalized into RealtimeEvents (string-based) for application consumers.

Purpose and Scope

This page documents the Realtime Engine & Event Model — the orchestration core and event contract of realtime-core:

• The RealtimeEngine class: construction, lifecycle, event routing, and the full public API (connect, disconnect, sendMessage, sendSignal, getHistory, markRead, startCall, endCall, onEvent).
• The event model: RealtimeEvent (domain shape), the wire SignalEvent shape (from messages.ts), and the normalization helpers toBigInt and evtToEvent in events.ts.
• The public package surface in index.ts, including the createRealtimeEngine factory and the exported types.

Intentionally left to sibling pages (referenced here but not deep-dived):

• Signal & Transport layer — SignalClient in signal-client.ts, the gRPC-Web transport in transport.ts, and message codecs in codec.ts implement the actual wire I/O; this page only covers the engine’s delegation to them.
• Media subsystem — MediaManager, MediaAdapter, and media session handling under media/; this page covers only the engine’s interception and forwarding of media_offer / media_answer / media_ice events.
• Message contracts — the exact payload shapes for MessageInput, HistoryInput, SendSignalInput and the SignalEvent wire type are defined in messages.ts and types.ts.

Overview

realtime-core is a headless realtime service — one RealtimeEngine instance per app session, covering both chats (text messages, history, read receipts over a signaling channel) and calls (media rooms over an SFU, negotiated via signaling). The engine is deliberately thin: it composes two subsystems and mediates the event flow between them:

1. Signaling side — a SignalClient maintains the connection to the realtime backend and emits raw events.
2. Media side — a MediaManager (backed by an injected MediaAdapter) handles WebRTC/SFU media sessions.

The engine’s core design decisions, visible directly in engine.ts:

• Single event funnel: every inbound event passes through the private route() method, which decides whether the event belongs to the media subsystem or to application listeners. This keeps media internals hidden from app code while giving the app one uniform onEvent subscription surface.
• Listener isolation: emit() wraps each listener callback in try/catch so a throwing subscriber can never break the event stream for other subscribers.
• Subscription as disposable: onEvent(cb) returns an unsubscribe function instead of requiring a separate API, so listeners are self-cleaning.
• Adapter injection for media: the engine accepts a MediaAdapter through EngineConfig.mediaAdapter; if no adapter is available, the engine runs in chat-only mode (media is null) and call APIs become no-ops.

The event model normalizes between two representations:

The mapper evtToEvent (in events.ts) performs this conversion, and toBigInt normalizes loose scalar inputs (strings, numbers, null, '', 0) to bigint — a pattern typical for gRPC-Web payloads where int64 values arrive as strings and absent values arrive as 0.

Architecture

Component roles:

• createRealtimeEngine(config) (index.ts) — the package factory. It is the only recommended construction path; it simply instantiates RealtimeEngine with the given EngineConfig.
• RealtimeEngine (engine.ts) — the orchestrator. Owns the SignalClient, the listener set, and (optionally) the MediaManager. All application-facing chat/call methods delegate to these collaborators.
• SignalClient (signal-client.ts) — the signaling connection. Constructed with a callback that the engine binds to route(), so every signal event enters the engine through one funnel.
• MediaManager (media/manager.ts) — the media session owner. It consumes media signaling events from the engine and drives the MediaAdapter; it reports outgoing signals back through the sendMedia callback, which the engine forwards as sendSignal({ targetId: 0, signalType, payload }).
• MediaAdapter (media/adapter.ts) — the pluggable WebRTC/SFU backend. defaultMediaAdapter() may return null, which disables media entirely.
• events.ts — the event-model boundary: toBigInt and evtToEvent convert between SignalEvent (wire) and RealtimeEvent (domain).

The direction of the arrows shows the one-way dependency discipline: the engine depends on signal and media collaborators; the signal client pushes events back into the engine; the engine never calls into the application’s listeners directly except through the isolated emit() loop.

Main Content: RealtimeEngine Implementation Walkthrough

Construction and Composition

The engine is created either directly or through the createRealtimeEngine factory. The constructor wires all collaborators in one place:

/** Headless realtime service — one instance per app session (chats + calls over gRPC-Web + SFU) */
export class RealtimeEngine {
  readonly signal: SignalClient;
  private listeners = new Set<(e: RealtimeEvent) => void>();
  private config: EngineConfig;
  private media: MediaManager | null;

  constructor(config: EngineConfig) {
    this.config = config;
    this.signal = new SignalClient(config, (e) => this.route(e));
    const adapter = config.mediaAdapter ?? defaultMediaAdapter();
    this.media = adapter ? new MediaManager(adapter, {
      sendMedia: (signalType, payload) => this.signal.sendSignal({ targetId: 0, signalType, payload: JSON.stringify(payload) }),
      onLocalStream: (s) => this.config.onLocalStream?.(s),
      onRemoteStream: (s) => this.config.onRemoteStream?.(s),
    }) : null;
  }

Source: engine.ts

Key mechanics:

1. Signal wiring: SignalClient is constructed with the engine’s config plus an event callback bound to this.route(e). From that point on, every event the signal layer produces is pushed through the engine’s private funnel — the engine never polls and never exposes the raw client callback.
2. Media resolution: config.mediaAdapter ?? defaultMediaAdapter() — an explicit adapter in config wins; otherwise the default adapter is used. The ?? treats an explicitly provided adapter as authoritative, so a caller can deliberately inject a custom WebRTC backend or stub.
3. Null-adapter degradation: adapter ? new MediaManager(...) : null — when no adapter exists, media is null and the engine silently runs in chat-only mode. startCall/endCall become no-ops (guarded with optional chaining), but messaging still works.
4. Media callbacks are bridged back to the engine:
   • sendMedia(signalType, payload) → this.signal.sendSignal({ targetId: 0, signalType, payload: JSON.stringify(payload) }). Media-layer signals are re-encoded as JSON strings and sent through the same signaling channel, targeting the SFU (targetId: 0 acts as the “room/server” target on the wire). This is why the app only ever deals with one outbound path.
   • onLocalStream / onRemoteStream are forwarded straight to the user-supplied EngineConfig callbacks, so the app observes media streams without touching MediaManager.

Event Routing: The Single Funnel

Every inbound event — chat or media — enters through route():

private route(e: RealtimeEvent): void {
  if (this.media && (e.eventType === 'media_offer' || e.eventType === 'media_answer' || e.eventType === 'media_ice')) {
    this.media.handleEvent(e.eventType, e.payload);
  } else {
    this.emit(e);
  }
}

Source: engine.ts

The routing decision is intentionally coarse: three event types are intercepted, everything else is forwarded. Media signaling events (media_offer, media_answer, media_ice — the classic SDP/ICE negotiation triplet) are consumed by MediaManager.handleEvent, which advances the WebRTC state machine internally. All other events (messages, presence, receipts, etc.) are emitted to application listeners. The this.media && guard means that in chat-only mode, even media events fall through to emit() — the app still sees them as opaque events rather than losing them.

Listener Management and Emission

private emit(e: RealtimeEvent): void {
  for (const listener of this.listeners) {
    try { listener(e); } catch { /* listener must not break the stream */ }
  }
}

onEvent(cb: (e: RealtimeEvent) => void): () => void {
  this.listeners.add(cb);
  return () => { this.listeners.delete(cb); };
}

Source: engine.ts

Design intent:

• The listener set is a plain Set, so onEvent is idempotent — registering the same callback twice results in one subscription, and unsubscribe is O(1).
• emit iterates the live set and isolates every listener in its own try/catch. A subscriber that throws cannot kill the stream for others; the comment in source makes the contract explicit: “listener must not break the stream”. Note the empty catch — the error is swallowed intentionally, trading diagnosability for resilience.
• onEvent returns a dispose closure capturing the callback. Consumers can attach it to component teardown (e.g., useEffect cleanup in React, onDestroy in frameworks) without holding a reference to the engine.

Public API: Delegation and Call Control

connect(): void { this.signal.connect(); }
disconnect(): void { this.signal.disconnect(); }

sendMessage(input: MessageInput): Promise<void> { return this.signal.sendMessage(input); }
sendSignal(input: SendSignalInput): Promise<void> { return this.signal.sendSignal(input); }
getHistory(input: HistoryInput) { return this.signal.getHistory(input); }
markRead(input: { peerId?: string | number; groupId?: string | number }): Promise<void> { return this.signal.markRead(input); }

/** Connect to an SFU media room and send the initial offer */
async startCall(roomId: string, kind: CallKind): Promise<void> {
  if (this.media) await this.media.connect(roomId, kind);
}

/** Leave the SFU media room */
endCall(): void {
  this.media?.end();
}

Source: engine.ts

Observations:

• Lifecycle (connect/disconnect) and chat operations (sendMessage, sendSignal, getHistory, markRead) are pure pass-throughs to SignalClient. The engine’s value here is a stable, minimal public surface: the app calls engine.sendMessage(...) and never touches transport concerns.
• markRead accepts either peerId or groupId — a 1 conversation or a group thread — reflecting that read receipts apply to both chat kinds. The fields are loosely typed (string | number) because IDs may arrive as strings on the wire but be constructed as numbers in app code; normalization happens downstream in events.ts / messages.ts.
• startCall is the only async call API: it awaits MediaManager.connect(roomId, kind) because the initial SDP offer must be generated before the method resolves. endCall is synchronous — tearing down the room has nothing to await.
• startCall returns undefined-promise when media is null (chat-only mode); callers should gate the call UI on whether an adapter is present.

Public Package Surface: index.ts

export { RealtimeEngine } from './engine';
export { SignalClient } from './signal-client';
export { toBigInt } from './events';
export * from './messages';
export type {
  CallKind,
  EngineConfig,
  RealtimeEvent,
  SendSignalInput,
  MessageInput,
  HistoryInput,
  MediaAdapter,
  MediaCallbacks,
} from './types';

import { RealtimeEngine } from './engine';
import type { EngineConfig } from './types';

export function createRealtimeEngine(config: EngineConfig): RealtimeEngine {
  return new RealtimeEngine(config);
}

Source: index.ts

The module deliberately exports classes, the event helper, and all message contracts — a complete SDK surface in one import. createRealtimeEngine is the sanctioned factory; exporting RealtimeEngine itself allows advanced use (subclassing, DI containers), while the factory keeps the common path one line: const engine = createRealtimeEngine(config).

Event Model: Wire Events → Domain Events

Normalization Helpers (events.ts)

export function toBigInt(v: string | number | bigint | undefined | null): bigint {
  if (v === undefined || v === null || v === '' || v === 0) return 0n;
  return BigInt(String(v));
}

export function evtToEvent(e: SignalEvent): RealtimeEvent {
  return {
    eventType: e.event_type,
    fromId: e.from_id.toString(),
    groupId: e.group_id === 0n ? undefined : e.group_id.toString(),
    payload: e.payload,
    createdAt: e.created_at.toString(),
  };
}

Source: events.ts

toBigInt(v) is the canonical scalar normalizer for gRPC-Web-style int64 fields. The four-way empty check (undefined | null | '' | 0) collapses every “absent” representation to 0n, then everything else is stringified before BigInt() — safe for numeric strings, number literals, and already-bigint values alike. This matters because gRPC-Web encodes int64 as decimal strings, and JSON round-trips can degrade them to numbers or null.

evtToEvent(e) is the wire→domain mapper and defines the event contract precisely:

The 0n → undefined rule for groupId is the key semantic: the wire uses 0 as a sentinel for “not in a group” (the same convention the engine reuses for the SFU target targetId: 0 in outbound media signals), while the domain model prefers optionality. Applications can then safely write if (event.groupId) without special-casing zero.

Event Routing Decision

The funnel shows how the this.media && guard combines with the type check: media interception requires both a live MediaManager and one of the three media event types. Otherwise the event is emitted. This is a deliberate degradation path — a chat-only engine still surfaces media events as opaque application events instead of swallowing them.

Core Flow: End-to-End Sequence

Walkthrough of the sequence:

1. Construction — createRealtimeEngine(config) builds the engine, which builds SignalClient (bound to route) and optionally MediaManager (bound to sendMedia/stream callbacks). No network I/O happens yet.
2. Subscription — onEvent(cb) registers the callback and returns the unsubscribe closure. Multiple subscribers are supported via the Set.
3. Connect — connect() delegates to SignalClient.connect(), opening the gRPC-Web channel.
4. Inbound events — the signal client asynchronously pushes normalized events into route(). Media events with a live media layer go to MediaManager.handleEvent; the manager may respond by invoking the sendMedia callback, which the engine forwards to sendSignal targeting the SFU (targetId: 0) — completing the signaling loop for negotiation. Non-media events go to emit() and reach every listener.
5. Outbound chat — sendMessage / getHistory / markRead go straight to the signal client; the app never sees the wire encoding.
6. Call lifecycle — startCall(roomId, kind) awaits the SFU room connection and initial offer; endCall() tears it down synchronously.
7. Disconnect — disconnect() closes the channel; listeners are not automatically cleared, so a re-connect() reuses the same subscription set.

Usage Examples

Basic Usage — Chat Session

The canonical setup: create the engine, subscribe, connect, send a message, and clean up.

import { createRealtimeEngine } from './index';
import type { EngineConfig } from './types';

const config: EngineConfig = {
  // signaling/server settings (see signal-client.ts and types.ts)
  // onLocalStream / onRemoteStream are optional media callbacks
};

const engine = createRealtimeEngine(config);

const unsubscribe = engine.onEvent((e) => {
  if (e.eventType === 'message') {
    console.log(`message from ${e.fromId}:`, e.payload);
  } else {
    console.log('event', e.eventType, e);
  }
});

engine.connect();
await engine.sendMessage({ /* MessageInput: peerId/groupId + content */ });

// later...
unsubscribe();
engine.disconnect();

Source: index.ts, engine.ts

This mirrors the engine’s own contract: onEvent returns a disposer, so teardown is symmetrical with setup. The fromId field is always a string and groupId is absent for 1 messages — exactly the normalization evtToEvent guarantees.

Call Flow — SFU Media Session

// Engine is configured with a media adapter (config.mediaAdapter or the default).
await engine.startCall('room-42', 'video');   // CallKind, e.g. 'audio' | 'video'
// ... media negotiation happens automatically via intercepted
//     media_offer / media_answer / media_ice events ...
engine.endCall();                              // leave the room synchronously

Source: engine.ts

The application never handles SDP or ICE manually — MediaManager consumes those events through route(), and outbound media signals flow back through sendSignal({ targetId: 0, signalType, payload }), which the media layer encodes as JSON strings. In chat-only mode (no adapter), startCall resolves without effect and endCall is a safe no-op due to optional chaining.

Normalizing IDs — BigInt to Domain String

import { toBigInt, evtToEvent } from './events';
import type { SignalEvent } from './messages';

// Wire values that all normalize to 0n:
toBigInt(undefined);   // 0n
toBigInt(null);        // 0n
toBigInt('');          // 0n
toBigInt(0);           // 0n
toBigInt('123');       // 123n
toBigInt(123);         // 123n

// Wire event → domain event:
const wire: SignalEvent = {
  event_type: 'message',
  from_id: 7n,
  group_id: 0n,        // "not in a group" sentinel
  created_at: 1700000000000n,
  payload: { text: 'hi' },
};
const evt = evtToEvent(wire);
// evt = { eventType: 'message', fromId: '7', groupId: undefined,
//         payload: { text: 'hi' }, createdAt: '1700000000000' }

Source: events.ts

This example makes the sentinel rule visible: group_id: 0n becomes groupId: undefined, while from_id and created_at always surface as strings. Applications that read event.groupId can rely on undefined rather than '0' for group-less events.

Configuration Options

EngineConfig is defined in types.ts (also passed to SignalClient, which consumes the connection-related fields). The options the engine itself reads, as evidenced in engine.ts:

Behavioral notes:

• config.mediaAdapter ?? defaultMediaAdapter() means an explicit falsy adapter (null/undefined assignment) defeats the default — the null-coalescing operator only falls back when the value is null/undefined, and if the fallback itself is null, media is disabled.
• Media callbacks in config are optional-chained (this.config.onLocalStream?.(s)), so an engine without stream handlers never throws when streams arrive.

API Reference

class RealtimeEngine — engine.ts

Constructor:

• constructor(config: EngineConfig) — builds the SignalClient (routing events through route) and the MediaManager (when an adapter resolves). Throws only if SignalClient construction fails on invalid config.

Members:

Properties:

• readonly signal: SignalClient — exposed for advanced use (direct signal access), though all normal operations route through the engine methods.

Event Types — types.ts / events.ts

RealtimeEvent (domain, consumed by listeners):

Helper functions:

• toBigInt(v: string | number | bigint | undefined | null): bigint — normalizes empty/absent scalars to 0n; otherwise BigInt(String(v)). Throws SyntaxError-style RangeError from BigInt() only if String(v) is not a valid integer string (e.g., 'abc'); '', null, undefined, and 0 are explicitly handled.
• evtToEvent(e: SignalEvent): RealtimeEvent — wire→domain mapping per the table above. Does not throw for well-formed SignalEvents.

Failure Modes, Edge Cases & Concurrency

Failure Modes

Edge Cases

• groupId: 0n sentinel — the wire’s 0n means “not in a group”; evtToEvent maps it to undefined, so if (event.groupId) is safe. The same zero-target convention appears outbound: media signals use targetId: 0 to address the SFU.
• Media events in chat-only mode — with media === null, media_offer/media_answer/media_ice fall through route() to emit(). A chat-only app that wants to observe negotiation (e.g., to log or forward it) can still do so.
• Duplicate subscriptions — Set semantics make onEvent(cb) idempotent; calling it twice with the same function registers once, and one unsubscribe removes it.
• fromId and createdAt as strings — the mapper stringifies bigint IDs and timestamps, so consumers never lose int64 precision to JavaScript Number coercion; they must parse createdAt explicitly if they need a Date.

Concurrency & Consistency

• Single-threaded event dispatch — emit() iterates the listener Set synchronously in the event loop. Since listeners run inline, a slow or blocking listener delays subsequent listeners and the signal client’s event processing. The try/catch protects against exceptions but not against slow handlers; keep callbacks lightweight or defer heavy work with queueMicrotask/setTimeout.
• One engine per session — the design contract (“one instance per app session”) means the listener Set and the signal connection are implicitly single-owner. Creating multiple engines per session would multiply signal connections; use one engine and fan out within the app instead.
• Call state machine isolation — media negotiation state lives inside MediaManager and is driven only via handleEvent; the engine never touches it, so the media state machine is not interleaved with chat event dispatch. Outbound media signals are serialized through the same sendSignal path as chat signals.
• Async boundary — startCall awaits MediaManager.connect (initial SDP offer), while endCall is synchronous; concurrent startCall/endCall ordering is the caller’s responsibility.

Performance & Operational Notes

• Listener fan-out is O(n) per event with no copy — emit iterates the live Set. With many subscribers, batch dispatch is cheap; with slow subscribers, it serializes the event loop (see concurrency note).
• No event buffering — events are delivered as they arrive; there is no queue or replay in the engine. Apps that need offline replay should rely on getHistory.
• Unsubscribe hygiene — each onEvent call allocates a closure; returning the disposer is designed for framework teardown, preventing listener leaks in long-lived sessions (SPAs, mobile runtimes).
• Media signaling overhead — media signals are JSON.stringify’d on the outbound path (sendMedia → sendSignal) and re-encoded on the wire; this is acceptable for negotiation messages (low frequency), not for media payloads themselves, which flow over the SFU data path.
• Connection lifecycle — connect/disconnect are direct delegates; reconnects and backoff are owned by SignalClient/transport.ts and are outside the engine’s scope.

Extension Points

1. Pluggable MediaAdapter — the primary extension seam. Inject a custom adapter via EngineConfig.mediaAdapter to replace the default WebRTC/SFU backend (e.g., a different SFU protocol, a simulator, or a test stub). Returning a falsy adapter yields chat-only mode.
2. EngineConfig.onLocalStream / onRemoteStream — media stream observation hooks; wire them to UI renderers (e.g., <video> element binding) without touching the engine.
3. onEvent subscription — the general-purpose extension point for any behavior: logging, analytics, forwarding to a store (Redux/Zustand), or re-emitting into an app-level event bus.
4. sendSignal passthrough — custom signaling payloads can be sent through the engine for app-specific handshakes alongside media negotiation.
5. RealtimeEngine export + factory — index.ts exports the class and the createRealtimeEngine factory; the class can be subclassed or DI-wrapped while the factory covers the common path.

Related Links

• Signal Client & Transport — the signaling connection the engine delegates to; owns reconnects and the gRPC-Web channel.
• Message Contracts — SignalEvent wire type and MessageInput/HistoryInput/SendSignalInput payload shapes.
• Media Manager — the SFU/WebRTC session owner consuming intercepted media events.
• Media Adapter — the pluggable media backend (defaultMediaAdapter) and MediaAdapter contract.
• Media Session — per-room session state driven by media_offer/media_answer/media_ice.
• Type Definitions — EngineConfig, RealtimeEvent, CallKind, and the exported public types.
• Package Entry — public exports and the createRealtimeEngine factory.
• Package Manifest — dependency and build configuration for realtime-core.