Realtime: server-sent events and WebSockets

// src/server/event-stream.server.ts
export type Send = (event: string, data: unknown, id?: string | number) => void;

export function eventStream(
  request: Request,
  start: (send: Send, close: () => void) => () => void,
  options: { keepAliveMs?: number; retryMs?: number } = {},
): Response {
  const encoder = new TextEncoder();
  let stopped = false;
  let cleanup: (() => void) | undefined;
  let keepAlive: ReturnType<typeof setInterval> | undefined;
  let controller!: ReadableStreamDefaultController<Uint8Array>;

  const write = (text: string) => {
    if (!stopped) controller.enqueue(encoder.encode(text));
  };
  const stop = () => {
    if (stopped) return;
    stopped = true;
    clearInterval(keepAlive);
    request.signal.removeEventListener("abort", onAbort);
    cleanup?.();
  };
  const close = () => {
    if (stopped) return;
    stop();
    controller.close();
  };
  const onAbort = () => close();

  const body = new ReadableStream<Uint8Array>({
    start(streamController) {
      controller = streamController;
      write(`retry: ${options.retryMs ?? 3000}\n\n`); // the browser's reconnection delay
      keepAlive = setInterval(() => write(": keep-alive\n\n"), options.keepAliveMs ?? 15_000);
      const send: Send = (event, data, id) => {
        const lines = [`event: ${event}`];
        if (id !== undefined) lines.push(`id: ${id}`);
        lines.push(`data: ${JSON.stringify(data)}`); // JSON has no raw newlines: one data line
        write(`${lines.join("\n")}\n\n`);
      };
      cleanup = start(send, close);
      if (stopped) cleanup(); // closed by `start` itself
      else if (request.signal.aborted) close();
      else request.signal.addEventListener("abort", onAbort, { once: true });
    },
    cancel() {
      stop();
    },
  });

  return new Response(body, {
    headers: { "content-type": "text/event-stream; charset=utf-8", "cache-control": "no-cache" },
  });
}

// src/routes/api/events/route.ts
import type { Endpoint } from "lumiose";
import { subscribe } from "../../../server/bus.server.ts";
import { eventStream } from "../../../server/event-stream.server.ts";

export const mode = "server";

export const GET: Endpoint<"/api/events"> = ({ request }) =>
  eventStream(request, (send) =>
    // The unsubscribe is the stream's cleanup: it runs when the client goes away.
    subscribe((message) => send("message", message, message.id)),
  );

// src/islands/LiveFeed.stsx (shortened), used as <LiveFeed client:load />
import { onCleanup, onMount, signal } from "lumiose";
import type { Message } from "../lib/messages.ts";

export function LiveFeed() {
  const messages = signal<Message[]>([]);
  const state = signal<"connecting" | "open" | "reconnecting">("connecting");

  onMount(() => {
    const source = new EventSource("/api/events");
    source.addEventListener("open", () => state.set("open"));
    // EventSource reconnects by itself, after the stream's `retry:` delay.
    source.addEventListener("error", () => state.set("reconnecting"));
    source.addEventListener("message", (event) => {
      const message = JSON.parse(event.data) as Message;
      messages.update((list) => [...list.slice(-49), message]);
    });
    // Closing the source ends the request, and the server's cleanup unsubscribes.
    onCleanup(() => source.close());
  });

  return (
    <ol>
      {messages().map((message) => (
        <li key={message.id}>{message.text}</li>
      ))}
    </ol>
  );
}

// src/server/chat-socket.server.ts (shortened)
import type { IncomingMessage } from "node:http";
import type { Duplex } from "node:stream";
import { WebSocket, WebSocketServer } from "ws";

const sockets = new WebSocketServer({ noServer: true, maxPayload: 2048 });

sockets.on("connection", (socket) => {
  socket.on("message", (data, isBinary) => {
    if (isBinary) return;
    const payload = JSON.stringify({ text: data.toString().slice(0, 500), at: new Date() });
    for (const client of sockets.clients) {
      if (client.readyState === WebSocket.OPEN) client.send(payload);
    }
  });
});

/** Upgrades requests to /ws and returns true; returns false for any other path. */
export function handleUpgrade(
  request: IncomingMessage,
  socket: Duplex,
  head: Buffer,
  origin: string | undefined = process.env.ORIGIN,
): boolean {
  const { pathname } = new URL(request.url ?? "/", "http://localhost");
  if (pathname !== "/ws") return false;
  // Browsers send the page's origin with the handshake, and the same-origin policy does not apply
  // to WebSockets: refuse other sites (cross-site WebSocket hijacking).
  if (!sameSite(request, origin)) {
    socket.end("HTTP/1.1 403 Forbidden\r\nConnection: close\r\nContent-Length: 0\r\n\r\n");
    return true;
  }
  sockets.handleUpgrade(request, socket, head, (ws) => sockets.emit("connection", ws, request));
  return true;
}

function sameSite(request: IncomingMessage, origin: string | undefined): boolean {
  const from = request.headers.origin;
  if (from === undefined) return true; // not a browser
  try {
    const host = origin === undefined ? request.headers.host : new URL(origin).host;
    return new URL(from).host === host;
  } catch {
    return false;
  }
}

// server.ts
import { createServer } from "@lumiose/adapter-node";
import { handleUpgrade } from "./src/server/chat-socket.server.ts";

const app = await createServer({ dir: new URL("dist", import.meta.url) });

app.server.on("upgrade", (request, socket, head) => {
  // Not the chat's path, and no other handler: refuse it.
  if (!handleUpgrade(request, socket, head)) socket.destroy();
});
console.log(`Listening on ${app.url}`);

// vite.config.ts
import type { Server } from "node:http";
import lumiose from "@lumiose/vite";
import { handleUpgrade } from "./src/server/chat-socket.server.ts";

function chatSockets() {
  return {
    name: "chat-sockets",
    configureServer(server: { httpServer: Server | null }) {
      server.httpServer?.on("upgrade", (request, socket, head) => {
        handleUpgrade(request, socket, head);
      });
    },
  };
}

export default { plugins: [lumiose(), chatSockets()] };

    location /ws {
        proxy_pass http://my_app;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        # nginx closes a connection that stays quiet for proxy_read_timeout (60 s by default).
        proxy_read_timeout 1h;
    }