@setu-ts/sse-plugin

Server-Sent Events — one-way, server → client streaming over text/event-stream. Registers an ISseService under CAPABILITIES.SSE ('sse').

Zero dependencies, built on IResponse.stream() and IRequestContext.signal.

When to use it

SSE runs over ordinary HTTP: it survives proxies and corporate middleboxes far more reliably than a socket upgrade, and browsers reconnect automatically. Reach for it whenever traffic only flows server → client — notifications, progress feeds, log tailing, metric ticks.

When the client also needs to send, use @setu-ts/websocket-plugin.

Installation

import { SsePlugin } from '@setu-ts/sse-plugin';

Usage

import { createApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { SsePlugin } from '@setu-ts/sse-plugin';
import { CAPABILITIES, type ISseService } from '@setu-ts/common';

const app = createApplication({
  plugins: [RuntimePlugin(), SsePlugin({ heartbeatMs: 15_000, retryMs: 3_000 })],
});
await app.start({ port: 3000 });

app.router.get('/events', async (ctx) => {
  const sse = ctx.services.get<ISseService>(CAPABILITIES.SSE);
  const conn = sse.open(ctx);
  conn.send({ id: '1', data: 'hello world' });
  return conn.result;
});

Named channels

const deploys = sse.channel('deploys');
deploys.publish({ data: JSON.stringify({ build: 412, status: 'live' }) });

channel(name) returns a reusable handle. Membership is allocated when a connection joins and reclaimed when the last member leaves. A held handle keeps working after reclamation: publishing resolves current members by name, including connections that joined through a later handle. Disconnect cleanup visits only the channels the connection joined.

channelCount, reported as channels by the health indicator, counts channels with members. peek(name) returns undefined for an empty channel and allocates nothing. Use it for presence reads:

app.router.get('/subscribers/:build', (ctx) => {
  const build = ctx.params.build ?? '';
  return ctx.response.json({ subscribers: sse.peek(`build:${build}`)?.size ?? 0 });
});

SseMessage.data is typed JsonValue — a recursive JSON-safe type, so the compiler admits exactly what the frame encoder can write. A bigint (which JSON.stringify throws on) and a function or symbol value (which it silently drops) are compile errors rather than runtime surprises. A property written T | undefined is fine: JSON.stringify drops the key. Three limits remain: a circular structure still throws at runtime; NaN, Infinity and -Infinity reach the wire as null, because JSON.stringify normalizes them rather than failing, so send the number as a string when that distinction matters; and a named interface does not assign, because TypeScript grants implicit index signatures only to object-literal types. Declare the payload with a type alias, or extend Record<string, JsonValue | undefined>.

Scaling beyond one replica

A channel is process-local until you register a backplane. On a single instance that is invisible. Behind two or more, channel('deploys').publish(...) reaches only the clients connected to that process — the other replicas’ subscribers hear nothing, and no error is raised anywhere. It is partial delivery, not a failure, which is what makes it easy to ship.

Registering @setu-ts/realtime-backplane-plugin with a cross-process transport is the entire fix. This plugin resolves it optionally, so nothing else changes:

import { RealtimeBackplanePlugin } from '@setu-ts/realtime-backplane-plugin';

createApplication({
  plugins: [
    RuntimePlugin(),
    RealtimeBackplanePlugin({ transport: 'redis', url: 'redis://localhost:6379' }),
    SsePlugin(),
  ],
});

The transport matters, not just the plugin. RealtimeBackplanePlugin() defaults to transport: 'memory' — a real bus, but a single-process one. Registering it bare silences the startup notice described below without fanning anything out. Use 'redis', or 'messaging' to reuse whichever broker MessagingPlugin already registered.

Channels now fan out across every replica sharing that transport. Remove the plugin and behavior returns to in-process, with no application change. channel.size stays local by design: it counts this replica’s subscribers, because a cluster-wide count is inherently asynchronous and cannot satisfy the synchronous getter.

When no backplane is registered this plugin logs one info line at startup stating the limitation. If you are running a single replica, that line is informational and safe to ignore — and if you have decided single-replica fan-out is correct for this deployment, scalingNotice: false silences it:

SsePlugin({ scalingNotice: false });

That suppresses the message only. Channel delivery is identical either way.

Options

Diagnostics (M98l)

SsePlugin({ diagnostics: { enabled: true, alias: 'feed' } }) counts, for the local diagnostics connector’s GET /v1/realtime (@setu-ts/diagnostics-plugin), how many streams opened and closed and how many frames the connection enqueued or refused — messages, comments and heartbeats alike; the initial retry: frame is written before the connection is attached and is not counted. A close caused by the 1 MiB backlog guard is failed and also counted in backpressureCloses. On each read of a paired devtool the source also reports the service’s current connectionCount and channelCount, which never expire the way operation records do. No message, comment, channel name, connection id or Last-Event-ID is ever captured.

Without the option the plugin registers an inert disabled source and nothing is attached, so its hot paths read no clock. Observation never changes a result: a refused send still throws the same error, and a failing clock or gauge reader latches the source collection-failed instead. A replacement registered later under the same capability is outside coverage and is never read. Enable only on an approved development dataset: counts aggregate every tenant using the instance.

Reconnection

A reconnecting browser sends Last-Event-ID. Set id on the messages you send so a client can resume from where it dropped.

Authentication

EventSource is the only browser API for this transport, and it can send exactly one credential: a cookie — no Authorization header, no custom header at all. So an EventSource behind requireAuth() is 401ed unless a strategy reads the cookie. The plugin needs no source change for this: SsePlugin.register registers no kernel route, the application owns the route, and SseService.open(ctx) receives the live IRequestContext — so once a cookie-bearing request produces a principal, requireAuth() on the route admits an EventSource exactly as it admits a fetch.

Configure the session arm on @setu-ts/auth-plugin and guard the route:

• SessionPlugin (from @setu-ts/session-plugin) registers the session middleware at priority 260.
• AuthPlugin({ session: { toPrincipal } }) — toPrincipal(view) maps the opened SessionView to the principal it carries, or returns null when the session holds no identity. Without SessionPlugin registered, AuthPlugin throws at register() naming both plugins.
• AuthPlugin registers the authentication middleware itself, at priority 300 — the authentication band — so no app.middleware.add(authMiddleware()) is needed.
• The route itself: app.router.get('/events', { middleware: [requireAuth()], handler }), where the handler opens the stream with sse.open(ctx).

The client must opt in to sending the cookie cross-origin: new EventSource('/events', { withCredentials: true }) — and the server must answer with a named origin plus Access-Control-Allow-Credentials: true, which corsMiddleware({ origin: 'https://example.com', credentials: true }) from @setu-ts/http-security-plugin does. origin: true (reflect any origin) cannot be combined with credentials: true; the middleware throws at registration, because that combination lets every site the user visits read credentialed responses.

The session cookie defaults to sameSite: 'lax', and an EventSource request is not a top-level navigation, so a Lax cookie is not sent on a cross-site EventSource request. Setting cookie.sameSite: 'none' on SessionPlugin removes that protection and exposes the stream to cross-site reads; if you must run none, the credentials CORS rule above is the minimum, and checking Origin in the authentication band is the stronger form.

Exports

Generated from the package barrel by pnpm run docs:exports; pnpm run check:docs fails when it drifts.

Full API

Every export and option is documented in PUBLIC_API.md.

API reference

SsePlugin() (@setu-ts/sse-plugin)

Provides Server-Sent Events (SSE) for real-time, one-way server-to-client messaging over text/event-stream. Built on the Milestone 42 IResponse.stream() primitive and IRequestContext.signal abort lifecycle.

Registration

import { SsePlugin } from '@setu-ts/sse-plugin';

app.register(
  SsePlugin({
    heartbeatMs: 15000,
    retryMs: 3000,
  }),
);

Usage

import { CAPABILITIES } from '@setu-ts/common';
import type { ISseService } from '@setu-ts/common';

app.router.get('/events', async (ctx) => {
  const sse = ctx.services.get<ISseService>(CAPABILITIES.SSE);
  const conn = sse.open(ctx);

  // Send immediately after handler returns — the stream stays open
  conn.send({ id: '1', data: 'hello' });

  return conn.result;
});

// Broadcast to a named channel
app.router.post('/broadcast', async (ctx) => {
  const sse = ctx.services.get<ISseService>(CAPABILITIES.SSE);
  sse.channel('updates').publish({ data: { msg: 'announcement' } });
  return ctx.response.json({ ok: true });
});

Options

Omitting an option disables that behaviour (no timer created).

Interface Reference

• ISseService.open(ctx): ISseConnection — opens a new SSE connection; sets headers, returns a connection with result (HandlerResult) the handler must return.
• ISseService.channel(name): SseChannel — returns a reusable named broadcast handle. Membership is allocated on add and reclaimed when the last member leaves. A held handle resolves current members by name, including members who joined after reclamation. Empty handles are not retained.
• ISseService.peek(name): SseChannel | undefined — the non-allocating counterpart. Returns the channel when it has members and undefined otherwise, registering nothing. Use it wherever the name comes from a request. Added in M74; a required member, so a replacement implementation of ISseService must supply it.
• ISseService.connectionCount: number — current open connections.
• ISseService.channelCount: number — channels with members, reported by the health indicator as channels. Drops when the last member leaves and becomes zero at shutdown. Disconnect cleanup visits only channels joined by that connection. Added in M74; a required member, like peek.
• ISseConnection.send(msg) — enqueue an encoded SSE frame (id:, event:, data: / multi-line data:, retry: + blank-line terminator).
• ISseConnection.comment(text) — enqueue a comment frame (: text\n\n).
• ISseConnection.close() — close the connection (idempotent).
• ISseConnection.lastEventId — the value of the Last-Event-ID request header (for resume logic).
• SseChannel.publish(msg) — broadcast to every open member, skipping closed ones.

Exports

Notes

• Built entirely on web-standard ReadableStream; no platform-specific server socket APIs.
• Channels are in-process until a backplane is registered. Register RealtimeBackplanePlugin and every publish also reaches members on other replicas; with no CAPABILITIES.REALTIME_BACKPLANE provider the behavior is unchanged. SseChannel.size keeps reporting local membership either way.
• SseMessage.data is JsonValue — a recursive JSON-safe type exported from @setu-ts/common. Until M74 the union was string | number | boolean | null | readonly unknown[] | Record<string, unknown>, whose last two arms admitted values JSON.stringify cannot represent, so this section’s own claim that the member “accepts any JSON-serializable value” was an aspiration rather than a guarantee. It is now enforced: a bigint (which throws) and a function or symbol value (which is silently dropped) are compile errors. A property written T | undefined still assigns, because JSON.stringify drops the key. Three limits remain and no type can close them: a circular structure throws at runtime; NaN, Infinity and -Infinity are members of number that JSON cannot represent, so JSON.stringify normalizes each to null and the value reaches the wire silently changed rather than refused (send it as a string when the distinction matters); and a named interface does not assign — TypeScript grants implicit index signatures only to object-literal types, which was true before the M70n widening and after it, so declare the payload with a type alias or extend Record<string, JsonValue | undefined>. What M70n’s widening actually bought was arrays, primitives and null. SseChannelImpl.publishLocal is the local-only delivery path the backplane subscriber uses; applications call publish.
• Cloudflare Workers and other edge platforms bound long-lived connections by their own limits — the plugin opens the stream the same way everywhere, but the platform may truncate the connection.
• The inject() method cannot read a streaming body and throws when it meets one; SSE integration tests must use a real socket (app.start({ port }) + fetch()).