realtime/sse_registry.ts

Generic registry of SSE streams for broadcasting to SSE clients.

Supports channel-based filtering — subscribers connect with optional channel filters, and broadcasts reach only matching subscribers.

Two identity slots enable both targeted disconnection and per-scope cap enforcement:

scope — a single capped identity (e.g., session hash). Subject to the per-scope cap and matched by close_by_identity. Use for the narrowest identity the subscriber belongs to.
groups — any number of uncapped identities (e.g., account id). Matched by close_by_identity but not subject to any cap. Use for coarser scopes a stream should be reachable by.

The split keeps "tabs-per-session" cap semantics sane when a stream also carries a broader identity for coarse close — the broader identity doesn't cap across sessions.

Registration can run in two phases (subscribe_pending then admit) so a revocation reaches a stream that is still opening — see the class doc.

view source

Declarations
#

5 declarations

PendingSubscription
#

realtime/sse_registry.ts view source

PendingSubscription import type {PendingSubscription} from '@fuzdev/fuz_app/realtime/sse_registry.js';

A registered subscription awaiting admission — what SseRegistry.subscribe_pending returns and SseRegistry.admit takes.

Its entry is already in the registry, so every close_by_identity can reach it, but nothing is delivered to it and it does not count toward the per-scope cap until it is admitted.

`unsubscribe`

Remove the registration, pending or admitted. Idempotent. Every path that refuses a pending subscription calls it, and an admitted stream's on_close listener does.

type (): void

returns void

SseRegistry
#

realtime/sse_registry.ts view source

import {SseRegistry} from '@fuzdev/fuz_app/realtime/sse_registry.js';

Generic registry of SSE streams with channel-based filtering and identity-keyed disconnection.

Subscribers connect with optional channel filters, a capped scope, and uncapped groups. Broadcasts go to a specific channel and reach only matching subscribers. close_by_identity force-closes all subscribers whose scope or groups contain the given key — use for auth revocation.

Two-phase registration

A stream is authorized once, at open, so a revocation has to be able to reach it from the moment its credential is read. A route's gates run before the stream exists, and each revocation closes only the streams *registered* under what it revoked — registering after the gates leaves a window in which a revocation closes nothing and the stream then opens on a dead session, or under a role its actor no longer holds.

So a route that authenticated a revocable credential registers in two phases: subscribe_pending registers the subscription un-admitted — closeable by every close_by_identity, and otherwise inert: it has no stream, so no broadcast reaches it, the cap neither counts nor evicts for it, and count does not see it (pending_count does). The route then re-reads the credential and the role, and admit attaches the stream and opens the feed — or refuses when a revocation closed the registration in between, evicting nothing. The cap's eviction happens at admission.

subscribe stays as the one-step form for a stream with nothing to revoke behind it. The twin of the Rust fuz_realtime SseRegistry (subscribe_pending / admit / subscribe).

Shutdown

close_all — the close AppServer.close runs — closes every stream, pending or admitted, and leaves the registry closing for good: every later registration is born closed. subscribe_pending registers nothing, so admit refuses the handle and the route's stream ends after its connect comment; subscribe closes the stream it is given at once. A stream still opening when the server shuts down therefore ends rather than outliving it. The flag is set in the same synchronous step that closes the live streams.

generics

SseRegistry<T>

`T`

examples

const registry = new SseRegistry<SseNotification>();

// subscriber connects (from SSE endpoint)
const unsubscribe = registry.subscribe(stream, {channels: ['runs']});

// when a run changes
registry.broadcast('runs', {method: 'run_created', params: {run}});

// subscriber disconnects
unsubscribe();

// scope = session hash (capped), groups = [account id] (close-only)
const unsubscribe = registry.subscribe(stream, {
  channels: ['audit_log'],
  scope: session_hash,
  groups: [account_id],
});

// coarse — close all of a user's streams on role revocation
registry.close_by_identity(account_id);

// fine — close just the stream(s) tied to a specific session
registry.close_by_identity(session_hash);

// two-phase, for a route behind a revocable credential
const pending = registry.subscribe_pending({scope: session_hash, groups: [account_id]});
if (!(await credential_is_still_live())) {
  pending.unsubscribe();
  return unauthorized();
}
const {response, stream} = create_sse_response(c, log);
// first: a client that already left closes the stream, its listener runs at
// once, and `admit` then refuses the removed registration
stream.on_close(pending.unsubscribe);
if (!registry.admit(pending, stream)) {
  stream.close(); // the body is the connect comment alone
} else {
  // expiry writes nothing a revocation closes on: end at the credential's deadline
  stream.on_close(
    schedule_credential_deadline(get_resolved_auth(c)?.credential_deadline ?? null, () =>
      stream.close()
    )
  );
}
return response;

`constructor`

type new <T>(options?: SseRegistryOptions | undefined): SseRegistry<T>

`options?`

type SseRegistryOptions

optional

`subscribe`

Add a subscriber, admitted at once — the one-step form of subscribe_pending + admit.

Not for a route that authenticated a revocable credential: a revocation landing between that route's gates and this call closes nothing, and the stream would open on a dead credential (class doc, "Two-phase registration").

Once the registry is closing, the subscriber is born closed: stream is closed at once, nothing is registered, and the returned function is a no-op.

type (stream: SseStream<T>, options?: SubscribeOptions | undefined): () => void

`stream`

SSE stream to send data to

type SseStream<T>

`options?`

channel filter and identity slots (scope + groups)

type SubscribeOptions

optional

returns () => void

unsubscribe function

mutates

registry — adds the new subscriber; closes oldest matching subscribers when `max_per_scope` is exceeded

`subscribe_pending`

The registration is closeable at once by every close_by_identity matching its scope or groups, and until admit it receives no broadcast, is not counted, and evicts nothing. A caller that then refuses the request calls the returned handle's unsubscribe.

Once the registry is closing, the registration is born closed: nothing is registered, and admit refuses the handle.

type (options?: SubscribeOptions | undefined): PendingSubscription

`options?`

channel filter and identity slots (scope + groups)

type SubscribeOptions

optional

returns PendingSubscription

the pending handle admit takes

mutates

registry — adds the un-admitted registration, unless the registry is closing

`admit`

Admit a pending subscription, attaching the stream it delivers to — phase two (class doc).

If the registration is gone — a revocation or the shutdown closed it while it was pending, it was born closed, or it was unsubscribed — nothing is evicted and the subscription is refused. Otherwise the per-scope cap is enforced as in subscribe, counting only admitted subscribers, and the stream starts receiving broadcast data. Admitting twice is refused the second time.

type (pending: PendingSubscription, stream: SseStream<T>): boolean

`pending`

the handle subscribe_pending returned

type PendingSubscription

`stream`

SSE stream to send data to

type SseStream<T>

returns boolean

false when the subscription was not admitted — the caller ends the stream

mutates

registry — attaches `stream`; closes oldest matching subscribers when `max_per_scope` is exceeded

`broadcast`

Broadcast data to all admitted subscribers on a channel.

Subscribers with no channel filter receive all broadcasts. Subscribers with a channel filter only receive matching broadcasts. A registration pending admission is not an open stream and is skipped — nothing is queued toward a subscriber not yet authorized.

type (channel: string, data: T): void

`channel`

type string

`data`

type T

returns void

`close_by_identity`

Force-close all subscribers whose scope or groups match the given key.

Closes each matching stream and removes the subscriber from the registry. Use for auth revocation — when a user's permissions change, close their SSE connections so they must reconnect and re-authenticate.

Matches every registration, pending or admitted — never filter on admission here: reaching a registration in flight is the point of the pending phase, and admit then refuses it.

A stream whose close throws does not stop the loop: every match is removed and closed, and the first error is thrown afterward.

type (identity: string): number

`identity`

the identity key to match (checked against scope and groups)

type string

returns number

the number of subscribers closed

throws

the - first error a stream's `close` threw, after every match was attempted

mutates

registry — removes matching subscribers and closes their streams

`close_all`

Force-close every subscriber, pending or admitted — the shutdown close AppServer.close runs through the registry's ConnectionCloser — and leave the registry closing, for good: every later registration is born closed (class doc, "Shutdown"). A second call closes nothing.

A stream whose close throws does not stop the loop: every subscriber is removed and closed, and the first error is thrown afterward.

type (): number

returns number

the number of subscribers closed

throws

the - first error a stream's `close` threw, after every subscriber was attempted

mutates

registry — sets the registry closing, removes every subscriber, and closes its stream

`count`

Number of open — admitted — subscribers. A registration still pending admission is counted by pending_count instead.

type number

getter

`pending_count`

Number of registrations awaiting admission — between subscribe_pending and admit (or their refusal). For telemetry and tests.

type number

getter

`closing`

Whether close_all has run — the registry is shutting down and every registration from now on is born closed (class doc, "Shutdown"). The audit stream route reads it to skip the re-reads for a registration admission will refuse anyway.

type boolean

getter

SseRegistryOptions
#

realtime/sse_registry.ts view source

SseRegistryOptions import type {SseRegistryOptions} from '@fuzdev/fuz_app/realtime/sse_registry.js';

Options for SseRegistry.

`max_per_scope?`

Max admitted subscribers sharing a single scope. When a subscriber is admitted and the count of admitted subscribers with the same scope has reached this limit, the oldest matching subscriber(s) are closed first. null (default) disables the cap. groups identities are never capped.

type number | null

`log?`

Logger for a stream whose close throws while the cap evicts it — the evicted stream is another subscriber's, so its failure is logged rather than failing the admission that evicted it. Unset falls back to an [sse] logger; null silences it.

type Logger | null

SubscribeOptions
#

realtime/sse_registry.ts view source

SubscribeOptions import type {SubscribeOptions} from '@fuzdev/fuz_app/realtime/sse_registry.js';

Options for SseRegistry.subscribe and subscribe_pending.

`channels?`

Channels to subscribe to. Empty/absent = all channels.

type readonly string[]

`scope?`

Primary (capped) identity — e.g., session hash. Subject to max_per_scope and matched by close_by_identity.

type string

`groups?`

Grouping identities — e.g., account id. Matched by close_by_identity but NOT subject to the cap. Use for coarse-targeted close.

type readonly string[]

Subscriber
#

realtime/sse_registry.ts view source

Subscriber<T> import type {Subscriber} from '@fuzdev/fuz_app/realtime/sse_registry.js';

One admitted subscriber, as SseRegistry.broadcast sees it.

generics

Subscriber<T>

`T`

`stream`

type SseStream<T>

`channels`

Channels this subscriber listens to. null means all channels.

type Set<string> | null

`scope`

Primary (capped) identity. null when none.

type string | null

`groups`

Grouping identities for close_by_identity. null when none.

type Set<string> | null

Depends on
#

realtime/sse.ts