@spfn/core/event

In-memory pub/sub event system with TypeBox-typed payloads, optional cache-backed multi-instance broadcast, job-queue fan-out, and Server-Sent Events (SSE) streaming to the browser. A WebSocket variant shares the same event definitions.

Import paths

There are four entry points. Picking the wrong one breaks the build (client code must never pull the Hono handler; the handler must never pull the EventSource client).

// Event definitions, router, route-map — isomorphic (define + emit + subscribe)
import { defineEvent, defineEventRouter, eventRouteMap } from '@spfn/core/event';

// Server SSE handler + token manager — SERVER ONLY (Hono, node:crypto)
import { createSSEHandler, SSETokenManager, CacheTokenStore } from '@spfn/core/event/sse';

// Browser SSE client — CLIENT ONLY (EventSource)
import { createSSEClient, createAuthSSEClient, subscribeToEvents } from '@spfn/core/event/sse/client';

// WebSocket router/handler/client — separate surface (see "WebSocket" below)
import { defineWSRouter } from '@spfn/core/event';          // or '@spfn/core/event/ws'
import { attachWSHandler } from '@spfn/core/event/ws';       // server
import { createWSClient } from '@spfn/core/event/ws/client'; // browser

createSSEClient / createAuthSSEClient / subscribeToEvents live in @spfn/core/event/sse/**client** — not in @spfn/core/event/sse. The bare @spfn/core/event/sse is the server handler surface.

In practice you rarely call createSSEHandler directly — defineServerConfig().events(router) mounts it for you (see Server setup).

Public API (complete)

From @spfn/core/event:

defineEvent(name) / defineEvent(name, schema) — define an event
defineEventRouter(events) — group events for SSE
defineWSRouter({ events, messages? }) — group events + message handlers for WebSocket
eventRouteMap — { eventsToken: { method: 'POST', path: '/events/token' } } (merge into RPC proxy)
.contract({ since, deprecatedIn?, removedIn? }) on an event and .contract({ auth, streamPath? }) on an event router — publish the event in contracts/current.json (see Contracted events)
.canSubscribe(predicate), .filter(predicate), .initial(producer), .grant(subject), .revoke(subject) on an event — who may subscribe over SSE, which frames a connection receives, the frames it receives right after connected, and changing an open connection (see Per-event permission)
Types: EventDef, EventHandler, EventSubscribeOptions, InferEventPayload, PubSubCache, JobQueueSender, EventContract, EventRouterContract, SubscribePredicate, SubscribeContext, SubscribeRequest, FramePredicate, FrameContext, InitialProducer, EventRouterDef, InferEventNames, InferEventPayloads, InferRouterEventPayload (the two-arg router variant — see note)

From @spfn/core/event/sse:

createSSEHandler(router, config?, tokenManager?) — Hono GET handler
SSETokenManager (class), CacheTokenStore (class)
Types: SSEToken, SSETokenStore, SSETokenManagerConfig, SSEMessage, SSEHandlerConfig, SSEHandlerAuthConfig, SSEAuthConfig, SSEClientConfig, SSEEventHandler, SSEEventHandlers, SSESubscribeOptions, SSESubscriptionChange, SSEConnectionState, SSEUnsubscribe, ConnectedPayload

From @spfn/core/event/sse/client:

createSSEClient(config?), createAuthSSEClient(config?), subscribeToEvents(events, handlers, config?)
Type: SSEClient, AuthSSEClientConfig

From @spfn/core/event/ws / @spfn/core/event/ws/client: see WebSocket.

Two InferEventPayloads. The name InferEventPayload exported from @spfn/core/event is the single-arg one (InferEventPayload<typeof userCreated> → payload of one event). The router module's two-arg version (<Router, 'eventName'>) is re-exported under the alias InferRouterEventPayload to avoid a clash. Inside the SSE types it is the two-arg form.

No such API. There is no event.on(...), no event.publish(...), no createEventBus(), no emitSync(). The model is: defineEvent → .subscribe() / .emit() / .useCache(). SSE clients are created with createSSEClient / createAuthSSEClient, never new EventSource directly (that loses typing and token handling).

Quick Start

import { defineEvent } from '@spfn/core/event';
import { Type } from '@sinclair/typebox';

// 1. Define an event with a typed payload
export const userCreated = defineEvent('user.created', Type.Object({
    userId: Type.String(),
    email: Type.String(),
}));

// 2. Subscribe (in-memory) — returns an unsubscribe function
const unsubscribe = userCreated.subscribe((payload) =>
{
    // payload: { userId: string; email: string }
    console.log('User created:', payload.userId);
});

// 3. Emit — awaits all handlers
await userCreated.emit({ userId: '123', email: 'user@example.com' });

// 4. Cleanup
unsubscribe();

Event without payload

export const serverStarted = defineEvent('server.started'); // EventDef<void>

serverStarted.subscribe(() => console.log('started'));
await serverStarted.emit();   // emit() takes no argument when there is no schema

emit is typed off the payload: () => Promise<void> for void events, (payload) => Promise<void> otherwise. Handlers run via Promise.allSettled — see error isolation.

EventDef methods

defineEvent returns an EventDef<TPayload>:

Member	Signature	Notes
`name`	`readonly string`	event name (e.g. `'user.created'`)
`schema`	`readonly TSchema?`	the TypeBox schema, if provided
`subscribe(handler, options?)`	`(payload) => void \| Promise<void>`, `{ localOnly?: boolean }` ⇒ returns `() => void`	in-memory; returns unsubscribe. `localOnly`: runs for this process's `emit` only, never for a cache delivery from another pod
`unsubscribeAll()`	`() => void`	drop all in-memory handlers, `localOnly` ones included
`emit(payload?)`	`Promise<void>`	fan-out to handlers (or cache) + job queues
`useCache(cache)`	`(PubSubCache) => Promise<EventDef>`	enable cross-instance broadcast (must `await`)
`contract(contract)`	`(EventContract) => EventDef`	publish in the contract; returns the same event; a second call throws
`canSubscribe(predicate)`	`<P>(SubscribePredicate<P>) => EventDef`	who may subscribe over SSE (once per connection); a second call throws
`filter(predicate)`	`(FramePredicate<TPayload>) => EventDef`	which frames a connection receives (sync, per frame); a second call throws
`initial(producer)`	`<P>(InitialProducer<TPayload, P>) => EventDef`	the frames a connection receives right after `connected`, before any live frame (once per connection with a subject); a second call throws
`grant(subject, match?)`	`Promise<void>`	activate the event on the subject's open SSE connections (those matching `match`, when given) where it is denied
`revoke(subject, match?)`	`Promise<void>`	deactivate the event on the subject's open SSE connections (those matching `match`, when given) where it is allowed

_registerJobQueue, _registerSubscriptionControl and _payload exist on the interface but are internal — _registerJobQueue is called by @spfn/core/job when a job does .on(event); _registerSubscriptionControl by createSSEHandler so grant / revoke reach its connections; _payload is a type-inference carrier (always undefined at runtime). Don't call them.

Server setup (via `defineServerConfig`)

To stream events to the browser, group them with defineEventRouter and register the router on the server. .events() mounts the SSE handler — you don't call createSSEHandler yourself.

// src/server/events.ts
import { defineEvent, defineEventRouter } from '@spfn/core/event';
import { Type } from '@sinclair/typebox';

export const userCreated = defineEvent('user.created', Type.Object({
    userId: Type.String(),
    email: Type.String(),
}));
export const orderPlaced = defineEvent('order.placed', Type.Object({
    orderId: Type.String(),
    userId: Type.String(),
    amount: Type.Number(),
}));

export const eventRouter = defineEventRouter({ userCreated, orderPlaced });
export type EventRouter = typeof eventRouter;

// server.config.ts
import { defineServerConfig } from '@spfn/core/server';
import { eventRouter } from './server/events';

export default defineServerConfig()
    .routes(appRouter)
    .jobs(jobRouter)
    .events(eventRouter)              // → GET /events/stream
    .build();

// Custom path + ping interval:
.events(eventRouter, {
    path: '/sse',                     // default: '/events/stream'
    pingInterval: 10000,              // keep-alive interval, ms (default: 10000 — under proxy idle timeouts)
})

.events(router, config) accepts Omit<SSEHandlerConfig, 'auth'> & { path?, auth?: SSEAuthConfig<TRouter> }. The auth field is the generic SSEAuthConfig<TRouter> so authorize/filter get full event-name and payload inference from your router.

config also carries the cross-pod transport knobs (see Multi-instance broadcast):

Option	Type	Default	Description
`multiInstance`	`boolean`	`true`	auto-wire Redis pub/sub when a cache is configured; `false` forces in-process
`channelPrefix`	`string`	env `SPFN_SSE_CHANNEL_PREFIX` or `spfn:sse:`	pub/sub channel prefix; set distinct prefixes to isolate apps sharing one Redis

.websockets(router, config) accepts the same multiInstance / channelPrefix knobs. Events shared by both routers are wired once.

Backpressure (maxQueue, default 1000). Each connection's outbound frames go through a single drain loop that awaits every write, so a slow client applies real backpressure instead of letting frames buffer unboundedly in memory. If the per-connection queue exceeds maxQueue, the connection is closed (the client reconnects) rather than dropping frames — dropping a chunk would corrupt an ordered token stream. Raise maxQueue for very bursty producers; lower it to cap memory more aggressively.

SSEHandlerConfig also declares a headers field, but the current handler only reads pingInterval, auth, and maxQueue. Setting custom headers here is a no-op — set response headers in middleware instead.

Contracted events (for separately compiled clients)

A web client takes its event names and payload types from typeof eventRouter. A native app cannot, so it hand-writes both — and they break at runtime. Declaring .contract() puts the events into contracts/current.json, where the contract gate refuses a breaking change at build:

export const sessionActivity = defineEvent('session.activity', Type.Object({
    sessionId: Type.String(),
})).contract({ since: '1.0.0' });            // deprecatedIn?, removedIn? as on routes

export const eventRouter = defineEventRouter({ sessionActivity, userCreated })
    .contract({ auth: 'tokenExchange' });     // streamPath?: string, default '/events/stream'

Only events carrying .contract() are published (userCreated above is not). The published name is the router key (sessionActivity), the name on the wire — not 'session.activity'.
.contract() on an event returns the same event and may be called once; a second call throws.
The router's auth is required: 'tokenExchange' when .events() enables auth, else 'none'. A streamPath not starting with / throws where it is declared.
Point the contract generator at the file with eventRouterPath — see the contract README.

Boot check. When the event router declares .contract(), createServer compares it with .events() before anything is served and refuses to start on a mismatch: the contract's streamPath (default /events/stream) must equal .events({ path }) (same default), and auth: 'tokenExchange' must match auth.enabled: true. A router without .contract() is not checked.

SSE event router contract declares streamPath /sse but .events() registers /events/stream; the contract would describe a path the server does not serve

An authorize hook narrowing which events a connection receives is per-connection policy and not part of the contract.

SSE authentication (Token Exchange)

The browser EventSource API cannot send custom headers, so a Authorization: Bearer JWT can't ride along on the stream request. SPFN uses a token exchange: an authenticated POST /events/token mints a one-time, short-TTL token; the stream request carries it as ?token=....

Client                              Server
  │  POST /events/token (Bearer JWT) │  ← protected by config.middlewares (e.g. authenticate)
  │ ───────────────────────────────► │  getSubject(c) → issue one-time token
  │  ◄─────────────────────────────  │  { token: "..." }   (default TTL 30s)
  │  GET /events/stream?token=…&events=…
  │ ───────────────────────────────► │  verify+consume token (one-time)
  │                                   │  → authorize(subject, events)
  │                                   │  → resolvePrincipal(subject, request)   (once)
  │                                   │  → event.canSubscribe(ctx)              (once per event)
  │                                   │  → event.initial(ctx)                   (once per allowed event; live frames held)
  │  ◄═══════════════════════════════ │  SSE stream opens (connected: subscribed + denied + initial)
  │  event: <name> / data: {…}        │  ← each initial frame, then the held live frames
  │  event: <name> / data: {…}        │  ← filter(subject, payload) + event.filter(payload, ctx) per emission

Enable it with auth: { enabled: true }:

import { defineServerConfig } from '@spfn/core/server';
import { authenticate } from '@spfn/auth/server';

export default defineServerConfig()
    .middlewares([authenticate])          // protects POST /events/token
    .routes(appRouter)
    .events(eventRouter, {
        auth: { enabled: true },
    })
    .build();
// → POST /events/token   (mints token; subject = c.get('auth').userId by default)
// → GET  /events/stream?token=…&events=…

The token endpoint path is derived from the stream path: /events/stream → /events/token, /sse → /token. It is registered with your app's named middleware applied — both the server-level .middlewares([...]) and the router's .use([...]) — so whatever authenticates your routes also authenticates token issuance.

Token store (multi-instance)

One-time tokens are stored. With auth.enabled and no explicit store, the server auto-detects a cache at startup via getCache():

Environment	Store	How
No cache (`CACHE_URL` unset)	`InMemoryTokenStore` (Map)	automatic fallback
Cache connected	`CacheTokenStore` (Redis/Valkey, `SET EX` + `GETDEL`, prefix `sse:token:`)	auto-detected
Custom	your `SSETokenStore` impl	pass `auth.store`

Manual / custom store:

import { CacheTokenStore } from '@spfn/core/event/sse';
import { getCache } from '@spfn/core/cache';

.events(eventRouter, {
    auth: { enabled: true, store: new CacheTokenStore(getCache()!) },
})

import type { SSETokenStore, SSEToken } from '@spfn/core/event/sse';

class DynamoTokenStore implements SSETokenStore
{
    async set(token: string, data: SSEToken): Promise<void> { /* ... */ }
    async consume(token: string): Promise<SSEToken | null> { /* one-time get+delete */ }
    async cleanup(): Promise<void> { /* remove expired */ }
}

Sharing a token manager with `@spfn/auth`

If you already run @spfn/auth's one-time-token system, hand SSE the same manager via tokenManager instead of letting it create its own — both SSE and direct API calls then draw from one token pool. Use a lazy resolver because getOneTimeTokenManager() only exists after the auth lifecycle's afterInfrastructure runs:

import { createAuthLifecycle, getOneTimeTokenManager } from '@spfn/auth/server';

export default defineServerConfig()
    .lifecycle(createAuthLifecycle())
    .events(eventRouter, {
        auth: {
            enabled: true,
            tokenManager: () => getOneTimeTokenManager(), // resolved at server start
        },
    })
    .build();

When tokenManager is provided, store/tokenTtl are ignored (the external manager owns them).

Authorization hooks

Both hooks get full type inference from the router (no casting).

authorize — runs once on connect, decides which events the subject may subscribe to. Return the allowed subset; an empty array ⇒ 403.

.events(eventRouter, {
    auth: {
        enabled: true,
        authorize: async (subject, events) =>
        {
            // events: ('userCreated' | 'orderPlaced')[] — inferred
            const user = await usersRepository.findById(subject);
            return user.role === 'admin' ? events : events.filter(e => !e.startsWith('admin.'));
        },
    },
})

filter — runs per emission, decides whether a given payload goes to this subject. Return false to skip. Payload is typed per event.

.events(eventRouter, {
    auth: {
        enabled: true,
        filter: {
            // payload: { orderId; userId; amount } — inferred
            orderPlaced: (subject, payload) => payload.userId === subject,
            // userCreated: omitted → delivered to all authorized subscribers
        },
    },
})

Per-event permission (`canSubscribe`, `filter`, `initial`, `grant` / `revoke`)

The global hooks above are one function for the whole router. When the rule belongs to one event, declare it where the event is defined:

export const auditLogged = defineEvent('audit.logged', payload)
    .canSubscribe<AuthPrincipal>(({ subject, principal, request }) => principal?.role === 'admin');

export const runtimeChanged = defineEvent('runtime.changed', payload)
    .filter((payload, { subject, request }) => payload.workspaceId === request.query.workspaceId)
    .initial(async ({ request }) => [await runtimes.current(request.query.workspaceId)]);

canSubscribe(ctx) — once per connection per requested event, only when the connection has a subject. true allows; false or a throw denies (a throw is logged at warn with the event name). A denial never refuses the connection: an event denied here is listed in the connected frame's deniedEvents, and a connection whose every event is denied stays open with zero subscriptions, so a later grant can activate one. May be async.
filter(payload, ctx) — per frame, with or without a subject; it matches the payload against the connect request. It runs beside the global filter map — both must return true. Sync. A throw drops that frame (warn) and leaves the connection alive.
initial(ctx) — the event's current frames for one connection, written right after the connected frame and before any live frame, so a client that opens the stream needs no read of its own. It runs once per connection with a subject, only for an event canSubscribe allowed, with the same { subject, principal, request }; the producers of one connection run in parallel, and their frames are written in the order the events were requested. It returns an array of payloads ([] sends nothing but still counts as served). Its frames skip both filters — they were produced for this connection — so the producer must return only what this subject may see: neither the global filter nor .filter() runs on its frames. They number their messageIds like any frame. The connected frame's initial lists exactly the events whose producer returned; a producer that throws or rejects is logged at warn with the event name (never the error) and sends nothing, and the stream stays open. While the producers run, live frames — and grant / revoke frames and pings — are held, then written after the initial frames in arrival order; a revoke landing meanwhile still stops the subscription at once. A grant later never runs it. A connection without a subject, or with no allowed event that declares it, gets initial: [].
request in all three is a snapshot taken once at connect: { query, headers } with lowercased header names — never the live Hono context. It is the connect request as sent: it holds the cookie / authorization headers and the consumed one-time token in query, so the app must never log the snapshot.
Each is set once per event: an EventDef is a shared singleton, so a second call throws.

resolvePrincipal — why the stream has no AuthContext. The subject comes from the one-time token, and the stream is mounted without the app's middleware, so there is no request auth object to hand a predicate. auth.resolvePrincipal(subject, request) loads it instead, once per connection, after the subject is known and before any canSubscribe; every predicate of that connection receives the same value as principal (type it with .canSubscribe<P>()). Absent → principal: undefined. A throw refuses the connection with 403 { error: 'Principal could not be resolved' }.

.events(eventRouter, {
    auth: {
        enabled: true,
        resolvePrincipal: async (subject) => principalsRepository.findBySubject(subject),
    },
})

Order. authorize (an empty result is still 403) → canSubscribe of each event authorize kept → per frame, the global filter and the event's filter. An event authorize dropped is never asked and counts as denied too.

Cost. At connect: one predicate call per requested event (plus one resolvePrincipal), and one initial producer call per allowed event that declares one. Per frame: sync filters only — never storage. The handler sets no timeout on them: a hanging canSubscribe or resolvePrincipal holds the connect open, and a hanging initial holds the connected frame and every live frame behind it, so the app owns the timeout. A client that leaves meanwhile is forgotten at once; the late verdict opens nothing, and late initial frames are not written. The initial frames and the held live frames go through the connection's bounded queue like any frame: a held frame counts against maxQueue from the moment it is held, together with the frames already queued, and past it the connection is closed as a slow consumer — while the producers still run, too — and the client reconnects. A producer's duration thus counts as the client's slowness: frames that pile up while it runs count as frames the client has not read.

grant / revoke — changing an open connection.

await workspaceRuntime.revoke(userId);   // membership ended
await workspaceRuntime.grant(userId);    // membership restored

revoke(subject) — on every open connection of the subject where the event is allowed: drop the subscription and send a subscription.revoked frame { event } (the router key). grant(subject) — where it is denied: subscribe (the same code path as at connect, with the stored request snapshot) and send subscription.granted. Anything else is a no-op, and no connection resolves without error.
match: { query: Record<string, string> } narrows either call to the connections whose connect query carries every named field with exactly that value — string to string, over the request snapshot. A connection that declared no such field never matches, not even against ''. Without match, every connection of the subject is reached.
```
// the tab on w1 only; a tab of the same user on w2 stays denied
await workspaceRuntime.grant(userId, { query: { workspaceId: 'w1' } });
```
With the cache transport attached (CACHE_URL), the call applies on this pod and is published once on the control channel <channelPrefix>$control; every other pod applies it to its own connections, with the same match rule. grant / revoke reject a match whose query is empty or holds a non-string value (an undefined field would vanish in JSON and widen the call on every other pod) before applying or publishing it. A control message whose match is not an object with a non-empty string-to-string query is malformed and ignored. Without a cache it is local only.
Three rules:
1. Open connections only. Neither re-runs canSubscribe or authorize — they are an operator's statement about connections that exist now.
2. Permission lives in app data. The next connect asks canSubscribe afresh, so a missed call delays until the client reconnects; it never produces a wrong answer after. A call that lands while a connect is still in authorize / resolvePrincipal / canSubscribe is not missed: it is recorded and applied over the predicates' verdict when that connection opens (the last call wins; no frame — the connected frame already lists the result).
3. An event never requested cannot be granted. The connection only knows the events in its ?events= list — request every event the page may need, and let canSubscribe deny.
A frame already enqueued before a revoke may still be written; none after it is.
WebSocket routers are not covered.

subscription.granted and subscription.revoked are stream-owned frames like connected and ping: not in the contract, and reserved as router keys of contracted events. On the client:

client.subscribe({
    events: ['runtimeChanged'],
    handlers: { runtimeChanged: (p) => update(p) },
    onSubscriptionChange: ({ event, status }) =>
    {
        // event: 'runtimeChanged' (typed); status: 'granted' | 'revoked'
    },
});

The connected frame — always the first of a connection — carries { subscribedEvents, deniedEvents, initial, timestamp } (ConnectedPayload; initial is absent from a server that predates .initial()). onConnected receives it on every connection, the first and each reconnect, after onOpen:

client.subscribe({
    events: ['runtimeChanged'],
    handlers: { runtimeChanged: (p) => update(p) },
    onConnected: ({ deniedEvents, initial }) =>
    {
        // initial: the events whose current frames follow this one; read the others yourself
    },
});

SSE auth config (`SSEAuthConfig<TRouter>`)

Option	Type	Default	Description
`enabled`	`boolean`	`false`	turn on token auth
`tokenTtl`	`number`	`30000`	token TTL (ms)
`store`	`SSETokenStore`	auto (Cache → InMemory)	token storage backend
`tokenManager`	`SSETokenManager \| () => SSETokenManager`	—	reuse an external manager (lazy resolver recommended); overrides `store`/`tokenTtl`
`getSubject`	`(c: Context) => string \| null`	`c.get('auth')?.userId ?? null`	extract subject on the token endpoint
`authorize`	`(subject, events[]) => events[] \| Promise<…>`	—	subscription authorization (once on connect)
`filter`	`{ [event]?: (subject, payload) => boolean }`	—	per-event payload filter
`resolvePrincipal`	`(subject, request) => unknown \| Promise<unknown>`	—	principal for every `canSubscribe` (once per connection); a throw ⇒ `403`

Browser client

`createSSEClient(config?)` — no auth

import { createSSEClient } from '@spfn/core/event/sse/client';
import type { EventRouter } from '@/server/events';

// Defaults: NEXT_PUBLIC_SPFN_API_URL (or http://localhost:8790) + /events/stream
const client = createSSEClient<EventRouter>();

const unsubscribe = client.subscribe({
    events: ['userCreated', 'orderPlaced'], // names inferred from EventRouter
    handlers: {
        userCreated: (payload) => console.log('user', payload.userId),
        orderPlaced: (payload) => console.log('order', payload.orderId),
    },
    onOpen: () => console.log('connected'),
    onError: (err) => console.error(err),
    onReconnect: (attempt) => console.log('reconnect', attempt),
    onClose: () => console.log('closed for good'),
});

client.getState();  // 'connecting' | 'open' | 'closed' | 'error'
client.close();     // tear down the active connection
unsubscribe();      // same as close() for that subscription

SSEClientConfig:

Option	Type	Default	Description
`host`	`string`	`NEXT_PUBLIC_SPFN_API_URL` or `http://localhost:8790`	backend host
`pathname`	`string`	`/events/stream`	SSE endpoint path
`url`	`string`	—	deprecated; full URL overriding host+pathname
`reconnect`	`boolean`	`true`	auto-reconnect on drop
`reconnectDelay`	`number`	`3000`	delay between attempts (ms)
`maxReconnectAttempts`	`number`	`0`	`0` = infinite
`withCredentials`	`boolean`	`false`	send cookies with the EventSource request
`acquireToken`	`() => Promise<string>`	—	mint a token before each (re)connect; appended as `?token=…`

Connect-query fields (`query`)

subscribe({ …, query }) appends extra fields to the stream URL — the key a keyed state event is filtered on, say. Each entry goes after events= and before token=, both name and value encodeURIComponent-ed, and every reconnect reuses the same query. The server reads them as request.query in canSubscribe and filter.

client.subscribe({
    events: ['state.runtime.changed'],
    handlers: { 'state.runtime.changed': (frame) => apply(frame) },
    query: { workspaceId: 'w 1' },   // → /events/stream?events=state.runtime.changed&workspaceId=w%201
});

events and token are reserved — the stream reads them itself — and subscribe throws naming the field before anything connects. Without query the URL is exactly what it was. The query belongs to the subscription: to change it, subscribe again (a new subscription supersedes the client's previous one). @spfn/core/state/client does this when its set of subscribed keys changes.

`createAuthSSEClient(config?)` — with auth (recommended)

Wraps createSSEClient and wires acquireToken to fetch a one-time token from the RPC proxy automatically on every (re)connect.

import { createAuthSSEClient } from '@spfn/core/event/sse/client';
import type { EventRouter } from '@/server/events';

const client = createAuthSSEClient<EventRouter>(); // POSTs /api/rpc/eventsToken under the hood

client.subscribe({
    events: ['userCreated'],
    handlers: { userCreated: (p) => console.log(p) },
});

AuthSSEClientConfig = Omit<SSEClientConfig, 'acquireToken'> plus rpcBaseUrl (default /api/rpc). It requires eventRouteMap merged into your RPC proxy so eventsToken resolves to POST /events/token. That token request is a cookie-authenticated POST through the proxy, so @spfn/auth's CSRF rule applies to it: the client mirrors the readable spfn_csrf cookie family into the x-spfn-csrf header, the way the api client does, so the stream opens under SPFN_AUTH_CSRF=enforce too. A custom acquireToken passed to createSSEClient has to send that header itself (csrfHeaderValue and documentCookieEntries from @spfn/core/nextjs build it).

// app/api/rpc/[routeName]/route.ts
import '@spfn/auth/nextjs/api';
import { createRpcProxy } from '@spfn/core/nextjs/server';
import { eventRouteMap } from '@spfn/core/event';
import { routeMap } from '@/generated/route-map';

export const { GET, POST } = createRpcProxy({
    routeMap: { ...routeMap, ...eventRouteMap },
});

`subscribeToEvents(events, handlers, config?)` — one-shot helper

import { subscribeToEvents } from '@spfn/core/event/sse/client';
import type { EventRouter } from '@/server/events';

const unsubscribe = subscribeToEvents<EventRouter>(
    ['userCreated'],
    { userCreated: (payload) => console.log(payload) },
    { host: 'https://api.example.com' }, // optional SSEClientConfig
);

It creates a throwaway client and subscribes once. For auth, prefer createAuthSSEClient (this helper takes a plain SSEClientConfig, so you'd have to wire acquireToken yourself).

Multi-instance broadcast (cross-pod fan-out)

By default emit only reaches handlers in the same process. Across multiple pods, an emit on the pod that handles the chat POST must still reach the SSE stream pinned to a different pod — otherwise the stream silently stalls.

This is automatic. When a cache is configured (CACHE_URL / getCache() returns a client), .events(router) and .websockets(router) wire every event to a Redis/Valkey pub/sub transport at startup. No code change in your app — register the router as usual and multi-pod fan-out just works. Without a cache it is a no-op and events stay in-process, so a single pod (or local dev) behaves exactly as before.

export default defineServerConfig()
    .events(eventRouter)          // CACHE_URL set → cross-pod fan-out; unset → in-process
    .build();

// Force in-process even when a cache is present, or set an isolation prefix:
.events(eventRouter, {
    multiInstance: false,         // default true
    channelPrefix: 'my-app:',     // default: env SPFN_SSE_CHANNEL_PREFIX, else 'spfn:sse:'
})

Mechanics (from event.ts): once a cache is wired, emit calls cache.publish(name, payload) instead of triggering local handlers directly — local handlers (including the SSE stream on the same pod) fire when the message comes back through the pub/sub subscribe callback. A handler subscribed with { localOnly: true } is the exception: emit runs it directly on the emitting pod, and a delivery through the cache never does — use it for work that must happen once per emit, not once per pod. Without a cache it runs on every emit like any handler. Auth filter still runs per-subscriber on the receiving pod's SSE handler — the transport only fans out by event name. Job queues receive the payload on every emit regardless of cache.

Process-global. multiInstance and channelPrefix are resolved once per process (first .events()/.websockets() wiring wins); a second router can't change them. That's correct because one process serves one app. An event shared by both routers is wired once.

Payloads must be JSON-serializable when fan-out is on (they cross Redis as JSON). A Date arrives as an ISO string; a void event arrives as null on remote pods (vs undefined in-process). bigint/circular payloads can't serialize — see degrade below.

Ordering needs sequential await and healthy publishes. Per-channel order is preserved end-to-end only if the producer awaits each emit in turn (for (const c of chunks) await event.emit(c)). Fire-and-forget (chunks.forEach(c => event.emit(c))) races the publishes and can reorder a token stream. Also: a mid-stream publish failure delivers its event via the synchronous local fallback, which can land ahead of an earlier successful emit whose echo is still in flight — so a Redis blip mid-stream may reorder relative to in-flight emits. The transport adds no sequence number; if strict order matters across failures, carry one in the payload and reorder on the client.

Channel isolation & receive-side trust. The default prefix spfn:sse: is shared by every SPFN app, so apps sharing one Redis with a colliding event name would cross-talk. Set channelPrefix / SPFN_SSE_CHANNEL_PREFIX per app (the server logs a WARN at startup if you're on the default); separate CACHE_URLs per app are isolated already. Note the receiving pod does not re-validate a payload against the event schema before delivering it to filter/handlers — anything with Redis access can publish to a channel, so treat Redis access as a trust boundary (same level as the one-time-token store).

Degrade: SSE is lossy (at-most-once). If a publish can't reach Redis (a blip) or the payload can't serialize, the event is still delivered to this pod's own subscribers (local fallback) and logged; only remote pods miss it. The same local fallback covers the asymmetric case where the publish succeeds but this pod's subscriber socket is down (its echo — the only path to same-pod streams — wouldn't arrive). A sustained publish outage trips a short circuit breaker so emits fast-path to local instead of each paying the publish timeout. A publish never crashes emit or kills a stream. ioredis auto-reconnects and re-subscribes (autoResubscribe, default true). A per-event SUBSCRIBE failure at startup degrades that event to in-process (logged) rather than aborting boot. Same-pod delivery costs one Redis round-trip — for a single-replica deploy with a cache, prefer multiInstance: false to skip it.

Manual `useCache` (advanced)

You rarely need this — the server wires it for you. EventDef.useCache(cache) is the underlying primitive if you drive events outside defineServerConfig (e.g. a worker). The PubSubCache shape it expects:

import { getCache } from '@spfn/core/cache';

const cache = getCache();
if (cache)
{
    await userCreated.useCache({
        publish: async (channel, message) =>
        {
            await cache.publish(channel, JSON.stringify(message));
        },
        subscribe: async (channel, handler) =>
        {
            const sub = cache.duplicate();
            await sub.subscribe(channel);
            sub.on('message', (ch, msg) =>
            {
                if (ch === channel) handler(JSON.parse(msg));
            });
        },
    });
}

await userCreated.emit({ userId: '123', email: 'a@b.com' }); // broadcasts to all instances

await useCache before emitting — it must finish subscribing first. A direct second call on the same EventDef logs Cache already configured and is ignored. (Separately, the server's auto-wiring dedups by event name across the SSE and WS routers — tracked in the transport, so a shared event's useCache is invoked only once and the warn never fires.)

Job integration

A job subscribing to an event with .on(event) registers itself on the event's job queue; the event's emit then enqueues the payload to every such queue (in addition to in-memory handlers / cache broadcast).

import { defineEvent } from '@spfn/core/event';
import { job, defineJobRouter } from '@spfn/core/job';
import { Type } from '@sinclair/typebox';

export const orderPlaced = defineEvent('order.placed', Type.Object({
    orderId: Type.String(),
    userId: Type.String(),
}));

export const sendConfirmation = job('send-order-confirmation')
    .on(orderPlaced)                          // payload type inferred from the event
    .handler(async (payload) => emailService.send(payload.orderId));

export const updateInventory = job('update-inventory')
    .on(orderPlaced)
    .handler(async (payload) => inventoryService.reserve(payload.orderId));

export const jobRouter = defineJobRouter({ sendConfirmation, updateInventory });

await orderPlaced.emit({ orderId: 'ord-1', userId: 'u-1' }); // both jobs enqueue

One emit fans out to in-memory handlers, the SSE stream, and every subscribed job queue — fully decoupled.

Pitfalls & anti-patterns

Wrong import depth for the client. createSSEClient / createAuthSSEClient / subscribeToEvents are in @spfn/core/event/sse/**client**, not @spfn/core/event/sse. The bare /sse is the server (Hono) surface; importing it into the browser pulls server code.
useCache not awaited before emit. useCache is async (it subscribes to the channel). Emitting before it resolves can drop the very first cross-instance events. Always await event.useCache(...) first.
Calling useCache twice. A second call logs Cache already configured for event: … and is ignored. Configure the cache once per event instance (typically at startup).
Emitting before subscribing (in-memory). subscribe registers a handler set; an emit that runs before any subscribe simply has no handlers. Register handlers / jobs at startup, not lazily after the first emit.
auth.headers/SSEHandlerConfig.headers does nothing. The handler reads only pingInterval, auth, and maxQueue. Don't expect custom response headers from config.
A slow client is dropped, not buffered forever. Writes are bounded (maxQueue, default 1000): if a client can't keep up with a fast producer, its connection is closed instead of growing memory without limit. Expect occasional reconnects under heavy streaming to slow clients; that's the backpressure working, not a bug.
Don't new EventSource(...) by hand. You lose type inference, the connected/ping control-event handling, one-time-token reconnect, and StrictMode-safe teardown that the client implements. Use createSSEClient / createAuthSSEClient.
Token-auth + browser auto-retry. A one-time token is consumed on first use, so the browser's built-in EventSource retry would reconnect with a dead token. The client handles this: on error with acquireToken set, it closes and reconnects through its own path to mint a fresh token. Don't disable reconnect expecting native retry to work with auth.
Stream-time errors aren't subscription errors. authorize returning [] ⇒ 403 at connect; invalid event names ⇒ 400; missing/expired token ⇒ 401. onError on the client fires for transport-level drops, not for these HTTP rejections — check the network response when a connection never opens.
Subject default. Without a custom getSubject, the token endpoint reads c.get('auth')?.userId. If your auth middleware stores the user elsewhere, supply getSubject or token issuance returns 401.
One subscription per client. createSSEClient(...).subscribe(...) supersedes any previous subscription on that client (the prior connection is closed). For independent streams, create separate clients.
SSETokenManager keeps a cleanup timer. It runs an unref'd interval; the in-memory store needs it to expire tokens. If you construct a manager manually (outside the server), call destroy() on shutdown. The Redis-backed CacheTokenStore relies on TTL and its cleanup() is a no-op.

WebSocket (bidirectional variant)

For client→server messages in addition to server→client push, use a WS router. It reuses the same defineEvent definitions and shares the SSE token manager / auth model.

// server
import { defineWSRouter } from '@spfn/core/event';     // also re-exported from /event/ws
import { attachWSHandler } from '@spfn/core/event/ws';

export const wsRouter = defineWSRouter({
    events: { userUpdated, notification },               // server → client push
    messages: {                                          // client → server handlers
        ping: ({ ws }) => ws.send('pong', {}),
        'chat.send': ({ payload, subject }) => handleChat(payload, subject),
    },
});
export type WSRouter = typeof wsRouter;

// usually mounted via defineServerConfig().websockets(wsRouter)

// browser
import { createWSClient } from '@spfn/core/event/ws/client';
import type { WSRouter } from '@/server/ws';

const client = createWSClient<WSRouter>();
client.subscribe({ events: ['userUpdated'], handlers: { userUpdated: (p) => {} } });
client.send('ping', {});

WSRouterDef extends EventRouterDef with a messages map. Message handlers receive { payload, subject?, ws } (WSMessageContext). See @spfn/core/event/ws exports for WSHandlerConfig, WSAuthConfig, WSClientConfig, etc.

Resource limits & backpressure (`WSHandlerConfig`)

The WS server is hardened like the SSE path — a slow or dead client cannot exhaust memory or connection slots:

Option	Default	Effect
`maxPayload`	`1048576` (1 MiB)	Max inbound frame size; larger frames are rejected by `ws` before they are buffered/parsed.
`maxBufferedBytes`	`1048576` (1 MiB)	Backpressure cap — if a connection's outbound buffer (`bufferedAmount`) exceeds this, the connection is closed with `1013` instead of buffering more (no OOM from a slow consumer). The client reconnects and re-subscribes.
`maxConnections`	`10000`	Global concurrent-connection cap; connections beyond it are rejected with `1013`.
`maxConnectionsPerSubject`	`0` (unlimited)	Per-authenticated-subject connection cap.

Keep-alive also tracks pongs: a socket that doesn't answer a ping by the next pingInterval tick is terminate()d, so half-open connections (sleeping device, NAT drop) are reaped instead of lingering with their subscriptions and buffers.

Types reference

interface EventDef<TPayload = void> {
    readonly name: string;
    readonly schema?: TSchema;
    subscribe: (handler: EventHandler<TPayload>, options?: { localOnly?: boolean }) => () => void;
    unsubscribeAll: () => void;
    emit: TPayload extends void ? () => Promise<void> : (payload: TPayload) => Promise<void>;
    useCache: (cache: PubSubCache) => Promise<EventDef<TPayload>>;
    readonly _contract?: EventContract;
    contract: (contract: EventContract) => EventDef<TPayload>;
    readonly _canSubscribe?: SubscribePredicate<any>;
    canSubscribe: <P = unknown>(predicate: SubscribePredicate<P>) => EventDef<TPayload>;
    readonly _filter?: FramePredicate<TPayload>;
    filter: (predicate: FramePredicate<TPayload>) => EventDef<TPayload>;
    readonly _initial?: InitialProducer<TPayload, any>;
    initial: <P = unknown>(producer: InitialProducer<TPayload, P>) => EventDef<TPayload>;
    grant: (subject: string, match?: SubscriptionMatch) => Promise<void>;
    revoke: (subject: string, match?: SubscriptionMatch) => Promise<void>;
}

interface SubscriptionMatch { query: Record<string, string>; }
interface SubscribeRequest { query: Record<string, string>; headers: Record<string, string>; }
interface SubscribeContext<P = unknown> { subject: string; principal: P | undefined; request: SubscribeRequest; }
type SubscribePredicate<P = unknown> = (ctx: SubscribeContext<P>) => boolean | Promise<boolean>;
interface FrameContext { subject: string | undefined; request: SubscribeRequest; }
type FramePredicate<TPayload> = (payload: TPayload, ctx: FrameContext) => boolean;
type InitialProducer<TPayload, P = unknown> = (ctx: SubscribeContext<P>) => TPayload[] | Promise<TPayload[]>;
interface ConnectedPayload { subscribedEvents: string[]; deniedEvents: string[]; initial?: string[]; timestamp: number; }

interface EventContract { since: string; deprecatedIn?: string; removedIn?: string; }

type EventHandler<TPayload> = (payload: TPayload) => void | Promise<void>;
type InferEventPayload<TEvent> = TEvent extends EventDef<infer P> ? P : never; // single-arg

interface PubSubCache {
    publish(channel: string, message: unknown): Promise<void>;
    subscribe(channel: string, handler: (message: unknown) => void | Promise<void>): Promise<void>;
}
type JobQueueSender = (queueName: string, payload: unknown) => Promise<void>;

interface EventRouterDef<TEvents> {
    readonly events: TEvents;
    readonly eventNames: (keyof TEvents)[];
    readonly _types: { [K in keyof TEvents]: TEvents[K]['_payload'] };
    readonly _contract?: EventRouterContract;
    contract: (contract: EventRouterContract) => EventRouterDef<TEvents>;
}
interface EventRouterContract { auth: 'none' | 'tokenExchange'; streamPath?: string; }
type InferEventNames<T>     = /* keyof router events as string union */;
type InferEventPayloads<T>  = T['_types'];
// router two-arg payload, exported as `InferRouterEventPayload`:
type InferRouterEventPayload<T, K> = T['_types'][K];

interface SSEMessage<TEvent extends string = string, TPayload = unknown> {
    event: TEvent; data: TPayload; id?: string;
}
type SSEConnectionState = 'connecting' | 'open' | 'closed' | 'error';

interface SSEToken { token: string; subject: string; expiresAt: number; }
interface SSETokenStore {
    set(token: string, data: SSEToken): Promise<void>;
    consume(token: string): Promise<SSEToken | null>; // one-time get+delete
    cleanup(): Promise<void>;
}

@spfn/core/job — background jobs that subscribe to events via .on()
@spfn/core/cache — Redis/Valkey backing for multi-instance broadcast + token store
@spfn/core/server — defineServerConfig().events() / .websockets()
MDN: Server-Sent Events