realtime-clients

npm create @setu-ts my-app -- --template rest
cd my-app && npm install
npm run start:dev

A live SSE/WebSocket server driven by the SDK clients; the smoke check exercises resume, authentication and keep-alive. The smoke check needs Bun installed and runs Wrangler through npx.

Run from the repository root after installing dependencies:

pnpm --dir apps/realtime-clients run start 3400
pnpm --dir apps/realtime-clients run smoke

Read the example deep dive for the concept and source walkthrough.

Read, in order

• setu.config.ts
• driver.ts
• smoke.ts

import { createApplication } from '@setu-ts/kernel';
import type { IKernelApplication } from '@setu-ts/kernel';
import { RuntimePlugin } from '@setu-ts/runtime';
import { SsePlugin } from '@setu-ts/sse-plugin';
import { WebSocketPlugin } from '@setu-ts/websocket-plugin';
import { registerModules } from './src/modules/index.ts';
import { IdleSocketPlugin } from './src/plugins/idle-socket.plugin.ts';

/**
 * Builds the server every realtime-client runtime check drives, without starting it.
 *
 * The heartbeat and idle timings are short so a check finishes in milliseconds.
 */
export function createApp(): IKernelApplication {
  const app = createApplication({
    plugins: [
      RuntimePlugin(),
      SsePlugin({ heartbeatMs: 10, retryMs: 1, scalingNotice: false }),
      WebSocketPlugin({ heartbeatMs: 20, idleTimeoutMs: 60, scalingNotice: false }),
      IdleSocketPlugin(),
    ],
  });

  registerModules(app.router);

  return app;
}

// A relative source import, not `@setu-ts/sdk`: this file also runs under
// plain Node and Bun with type stripping, which do not resolve workspace paths.
import { createRealtimeClient, createSseClient } from '../../packages/sdk/src/index.ts';
import { IDLE_SOCKET_PATH, SMOKE_TOKEN } from './src/contract.ts';

const wait = (milliseconds: number): Promise<void> =>
  new Promise((resolve) => setTimeout(resolve, milliseconds));

/** Checks that the SSE client reconnects with `Last-Event-ID` and skips heartbeats. */
async function checkSseClient(baseUrl: string): Promise<void> {
  const received: Array<{ readonly id?: string }> = [];
  let twoEventsArrived: () => void = () => {};
  const complete = new Promise<void>((resolve) => {
    twoEventsArrived = resolve;
  });
  const sse = createSseClient({
    url: `${baseUrl}/events`,
    headers: { authorization: `Bearer ${SMOKE_TOKEN}` },
    reconnect: { delayMs: 1, maxAttempts: 3 },
    onEvent: (event) => {
      received.push(event);
      if (received.length === 2) twoEventsArrived();
    },
  });
  await Promise.race([
    complete,
    wait(2_000).then(() => Promise.reject(new Error('SSE reconnect did not deliver two events.'))),
  ]);
  sse.close();

  if (received.some((event) => event.id === undefined)) {
    throw new Error('An SSE heartbeat comment leaked as an event.');
  }
  const { resumedWith } = (await (await fetch(`${baseUrl}/resume`)).json()) as {
    resumedWith: string | null;
  };
  if (resumedWith !== received[0]?.id) {
    throw new Error('The SSE reconnect did not send Last-Event-ID.');
  }
}

/** Checks that a read-only WebSocket client survives the server's idle sweep. */
async function checkIdleSocket(baseUrl: string): Promise<void> {
  let closedUnexpectedly = false;
  let stopping = false;
  let opened: () => void = () => {};
  const open = new Promise<void>((resolve) => {
    opened = resolve;
  });
  const socket = createRealtimeClient({
    url: baseUrl.replace(/^http/, 'ws') + IDLE_SOCKET_PATH,
    parse: (data) => data,
    reconnect: { maxAttempts: 0 },
    onMessage: () => {},
    onStateChange: (state) => {
      if (state === 'open') opened();
      if (state === 'closed' && !stopping) closedUnexpectedly = true;
    },
  });
  await Promise.race([
    open,
    wait(2_000).then(() => Promise.reject(new Error('The WebSocket client did not connect.'))),
  ]);
  // Timed from the open, and longer than the server's idle timeout, so only
  // heartbeats can keep the socket alive.
  await wait(140);
  stopping = true;
  socket.close();
  if (closedUnexpectedly) {
    throw new Error('The read-only WebSocket client did not survive the idle sweep.');
  }
}

/** Drives both SDK clients against the real server, from Node, Bun or workerd. */
export async function exerciseRealtimeClients(baseUrl: string): Promise<void> {
  await checkSseClient(baseUrl);
  await checkIdleSocket(baseUrl);
}

if (import.meta.main) {
  const baseUrl = process.argv[2];
  if (baseUrl === undefined) throw new Error('Expected the realtime server URL.');
  await exerciseRealtimeClients(baseUrl);
}

import assert from 'node:assert/strict';
import { after, before, describe, it } from 'node:test';
import { run } from '../../scripts/lib/run.ts';
import { freePort, SOURCE_HOST_ARGS, startHost, stopHost, waitForHttp } from '../smoke-helpers.ts';
import type { Host } from '../smoke-helpers.ts';
import { exerciseRealtimeClients } from './driver.ts';

/** Runs a command to completion and fails, naming it, if it does. */
async function runOrFail(command: string, args: readonly string[]): Promise<void> {
  const outcome = await run(command, args, { inherit: true });
  if (outcome.code !== 0) throw new Error(`${command} ${args.join(' ')} exited ${outcome.code}.`);
}

describe('realtime-clients example', () => {
  let server: Host;
  let baseUrl: string;

  before(async () => {
    const port = await freePort();
    baseUrl = `http://127.0.0.1:${port}`;
    server = startHost(process.execPath, [...SOURCE_HOST_ARGS, 'main.ts', String(port)]);
    await waitForHttp(`${baseUrl}/resume`, { label: 'Realtime client server' });
  });

  after(async () => {
    assert.equal(await stopHost(server), 0, 'the server did not shut down cleanly');
  });

  it('works from this Node process', async () => {
    await exerciseRealtimeClients(baseUrl);
  });

  it('works from Node with type stripping', async () => {
    await runOrFail('node', ['--experimental-strip-types', 'driver.ts', baseUrl]);
  });

  it('works from Bun', async () => {
    await runOrFail('bun', ['driver.ts', baseUrl]);
  });

  it('works from workerd', async () => {
    // `npx wrangler` downloads Wrangler on first use, which on a cold runner
    // takes longer than the readiness budget below. Resolve it first.
    await runOrFail('npx', ['--yes', 'wrangler', '--version']);

    const workerdPort = await freePort();
    const worker = startHost('npx', [
      'wrangler',
      'dev',
      'worker.ts',
      '--port',
      String(workerdPort),
      '--log-level',
      'error',
    ]);
    try {
      const workerdUrl = `http://127.0.0.1:${workerdPort}`;
      await waitForHttp(workerdUrl, { label: 'workerd client host' });
      const response = await fetch(`${workerdUrl}/?baseUrl=${encodeURIComponent(baseUrl)}`);
      if (!response.ok) throw new Error(`The workerd client driver returned ${response.status}.`);
    } finally {
      await stopHost(worker);
    }
  });
});