Streams: server-sent events and WebSocket

A route file can export SSE (an event stream) or WS (a WebSocket), instead of a page or API methods. A file is one or the other. The path is the file's path, as for pages.

import type { RequestEvent, SseStream } from "@shadow-js/dusk/server";

export function SSE(event: RequestEvent, stream: SseStream): void {
  // `stream.lastEventId` is the Last-Event-ID the browser sent when it reconnected.
  void event;
  void stream.send({ id: "1", event: "price", data: { symbol: "ACME", value: 12.5 } });
}

The stream sends a heartbeat comment every 15 seconds, aborts stream.signal when the client disconnects, and closes a consumer that falls too far behind: send rejects with SseOverflowError once more than 256 sends wait (streams.sseMaxBufferedEvents). A producer that awaits each send is paced by the consumer instead. The heartbeat and stream.comment() are keep-alives, not data: they are written only while the consumer has read what was queued before (a client that never reads gets at most one waiting), and a comment is cut to 1 KiB. createEventLog keeps the last events, so a reconnecting client gets what it missed: log.since(stream.lastEventId) (undefined means the id is gone and the client must resync).

import type { WebSocketConnection, WebSocketHandlers } from "@shadow-js/dusk/server";

export const WS: WebSocketHandlers = {
  // Refuse the connection by returning a Response; here the request must carry a session cookie.
  upgrade(event) {
    if (event.cookies.get("session") === undefined) return new Response("sign in", { status: 401 });
    return undefined;
  },
  open(ws: WebSocketConnection) {
    ws.send("welcome");
  },
  message(ws: WebSocketConnection, message: string | Uint8Array) {
    ws.send(message);
  },
};

The upgrade is checked before the socket is handed over: the handshake (RFC 6455: version 13, a 16-byte key, Connection: Upgrade), then Origin (the request's own origin, or one in trustedOrigins; a missing Origin is refused unless security.csrf.allowMissingOrigin). Messages are limited to 1 MiB and 100 per second (1009 and 1008 when exceeded), the server pings every 30 seconds and drops a peer that does not pong within 10, and a protocol violation closes with its RFC code. Node serves WebSocket routes through the upgrade event of serve(); other runtimes call handler.upgrade(request) and write the 101 themselves. The app-wide middleware runs around the route's WS.upgrade hook, after the handshake and the origin check, and a response it returns refuses the upgrade; the hook is the place for the route's own authorization check, and it must not trust anything but the request. The whole decision (middleware and hook) is bounded by limits.requestTimeoutMs: one that never settles is a 504 and gives its connection slot back. A connection whose transport holds more than 8 times wsMaxMessageBytes of unsent bytes (a peer that stopped reading while the server sends) is dropped with 1013, and ws.send returns false from then on. ws.close(code, reason) cuts the reason to 123 UTF-8 bytes at a character boundary.

In the browser, createSseClient and createWebSocketClient (in @shadow-js/dusk/client) reconnect with backoff; a 4xx stops an SSE client, and close() is final. In a component, useSSE(url) and useWebSocket(url) return signals and close when the component is disposed.

import { createSseClient, createWebSocketClient } from "@shadow-js/dusk/client";

const prices = createSseClient<{ symbol: string; value: number }>({
  url: "/prices",
  onEvent: (event) => console.log(event.data.symbol, event.data.value),
});
const chat = createWebSocketClient({ url: "/chat", onMessage: (data) => console.log(data) });
chat.send("hello");
prices.close();
chat.close();