Ingestion & lifecycle

The data flow#

client getStats()  ──►  ClientSample  ──►  observer.accept(sample, ctx?)
                                               │
              ┌────────────────────────────────┘
              ▼
   accept middlewares (global, in order)
              │
              ▼
   get-or-create ObservedCall ──► get-or-create ObservedClient ──► client.accept(sample, ctx)
                                                                        │
                                              per peerConnections[] in the sample
                                                                        ▼
                                              get-or-create ObservedPeerConnection
                                              .accept(pcSample, ctx) updates all sub-stats,
                                              derives deltas/bitrates/RTT, correlates remote RTP
                                                                        │
                          metrics roll up: PeerConnection → Client → Call → Observer
                                                                        │
                                          events emitted on the Observer bus

observer.accept(sample, context?)#

The single entry point. It:

1. drops the sample and emits sample-rejected if the observer is closed;
2. runs it through the global accept-middleware chain;
3. drops it and emits sample-rejected if callId or clientId is missing;
4. gets or lazily creates the ObservedCall and ObservedClient;
5. delegates to client.accept(sample, context), which fans out to each peer connection.

Sub-entities that stop appearing in samples are garbage-collected by a “visited” mark-and-sweep on each ObservedPeerConnection.accept(), emitting the corresponding *-removed events.

Accept middlewares#

observer.addAcceptMiddleware(...) registers middlewares run on every sample inside accept(), in order, before the sample is dispatched to any call or client. Each gets a { sample, context } payload; it can inspect or mutate either, then call next(payload) to continue.

import { Observer, AcceptMiddleware } from '@observertc/observer-js';

const observer = new Observer();

// Derive callId/clientId from the app's own attachment, before dispatch.
const route: AcceptMiddleware = ({ sample }, next) => {
    sample.callId ??= sample.attachments?.roomId as string;
    sample.clientId ??= sample.attachments?.peerId as string;
    next({ sample });
};

// Drop samples from a blocklisted client — never dispatched.
const filter: AcceptMiddleware = (payload, next) => {
    if (blocked.has(payload.sample.clientId)) return;   // no next() => dropped
    next(payload);
};

observer.addAcceptMiddleware(route, filter);
observer.removeAcceptMiddleware(route);

Not calling next drops the sample — nothing is created and no event fires. A throwing middleware is caught and warns (the sample is dropped) rather than crashing accept(). When no middleware is registered, accept() dispatches directly with no overhead.

context versus appData#

type AcceptContext = Record<string, unknown>;

A single optional object threaded down the whole accept chain. It is transient request-scoped data, never written to appData and not stored on any entity.

client-updated and peer-connection-updated carry the exact context of that sample; call-updated carries the context of the client accept() that drove the call update (absent for interval- or teardown-driven updates).

appData factories#

Instead of pre-creating an entity just to enrich its appData, register a factory once. It runs whenever the entity is created without an explicit settings.appData — including lazily inside accept() — and receives the accept context that caused the creation:

const observer = new Observer({
    createCallAppData:   ({ callId, acceptCtx })      => ({ callId, startedAt: Date.now(), tenant: acceptCtx?.tenant }),
    createClientAppData: ({ clientId, observedCall }) => ({ clientId, tenant: observedCall.appData.tenant }),
});

observer.accept(sample, { tenant: 'acme' });

That is what lets an accept middleware resolve something once — a tenant, a trace id — and have it land in appData at birth, instead of every factory re-deriving it from the sample. appData stays application-owned: the context is offered to the factory, never written across by the library.

The client factory receives the already-created parent observedCall, so it can derive fields from it.

Get-or-create helpers#

const call = observer.getOrCreateObservedCall({ callId, appData }, acceptCtx?);
const client = call?.getOrCreateObservedClient({ clientId, appData }, acceptCtx?);

These return undefined (and warn) when the parent is closed. createObservedCall / createObservedClient return the existing instance (and warn) if the id already exists.

When things update#

“Update” means recompute aggregated metrics, run the detectors, and emit the *-updated event at that level. Updates are event-driven — there is no built-in timer.

A call is updated when any of its clients is updated. The observer is updated when any of its calls is updated. Composed: the observer is updated exactly when any client anywhere is updated.

const observer = new Observer({ autoUpdateOnCallUpdate: false });
setInterval(() => observer.update(), 5_000);

The updatePolicy / defaultCallUpdatePolicy enum and the pluggable Updater are gone in 1.0. “When all clients have updated” sounds appealing and deadlocks on the first client that stops sending — one silent participant froze the whole call’s aggregation until it timed out.

Automatic teardown#

new Observer({
    closeClientIfIdleForMs: 60_000,   // a client with no sample for this long auto-closes
    closeCallIfEmptyForMs: 20_000,    // a call with zero clients for this long auto-closes
});

Closing cascades down (call → clients → peer connections → sub-stats), unsubscribing listeners and emitting the *-closed / *-removed events.

Issues still open when a client closes are force-resolved with resolvedBy: 'client-closed', and the registry additionally expires stale entries — so a crashed participant cannot leave an issue “active” forever.

Error-handling philosophy#

The library warns and degrades; it does not throw on operational problems:

sample-rejected carries reason: 'observer-closed' | 'missing-callId' | 'missing-clientId' and the sample itself, so a rejected sample is observable rather than silent.