Realtime backplane example

npm create @setu-ts my-app -- --example realtime
cd my-app && npm install
npm start

Each process creates one replica using the Redis transport (never the process-local memory transport).

REDIS_URL=redis://127.0.0.1:6379 pnpm --dir apps/realtime run start 3000
REDIS_URL=redis://127.0.0.1:6379 pnpm --dir apps/realtime run start 3001

The smoke task starts two replicas as separate processes, subscribes an SSE client to B, then proves a publish to A reaches that client. It reports a skip only when REDIS_URL is unavailable.

The separate processes are what makes the check mean anything: run both replicas inside one process and the backplane’s process-local 'memory' transport delivers the message on its own, so the check passes without any cross-replica delivery. Swapping this example to transport: 'memory' must make pnpm --dir apps/realtime run smoke fail.

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

Read, in order

• src/modules/news/index.ts
• src/modules/news/news.controller.ts
• smoke.ts

import type { IRouterApi } from '@setu-ts/common';
import { registerNewsRoutes } from './news.controller.ts';

/** Registers the news feature. */
export function registerNewsModule(router: IRouterApi): void {
  registerNewsRoutes(router);
}

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

/** The SSE channel every replica's subscribers join. */
export const NEWS_CHANNEL = 'news';

/**
 * Registers the news stream and its publish route.
 *
 * A message published on any replica reaches subscribers on every replica,
 * because the channel publishes through the realtime backplane.
 */
export function registerNewsRoutes(router: IRouterApi): void {
  router.get('/events', (ctx) => {
    const sse = ctx.services.get<ISseService>(CAPABILITIES.SSE);
    const connection = sse.open(ctx);
    sse.channel(NEWS_CHANNEL).add(connection);
    return connection.result;
  });

  router.post('/publish', async (ctx) => {
    const { message } = await ctx.request.json<{ message: string }>();
    ctx.services
      .get<ISseService>(CAPABILITIES.SSE)
      .channel(NEWS_CHANNEL)
      .publish({ data: message });
    return ctx.response.status(204).send();
  });
}

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

const redisUrl = process.env['REDIS_URL'] ?? skip('set REDIS_URL to run the two-replica check.');

interface Replica {
  readonly host: Host;
  readonly baseUrl: string;
}

/**
 * Starts one replica in a process of its own.
 *
 * Two replicas in one process would share the backplane's in-process
 * transport, so the check would pass with no cross-replica delivery at all.
 */
function startReplica(port: number): Replica {
  const host = startHost(process.execPath, [...SOURCE_HOST_ARGS, 'main.ts', String(port)], {
    ...process.env,
    REDIS_URL: redisUrl,
  });
  return { host, baseUrl: `http://127.0.0.1:${port}` };
}

/** Reads an SSE stream until a `data:` line carrying the expected message arrives. */
async function readUntil(stream: ReadableStream<Uint8Array>, expected: string): Promise<void> {
  const decoder = new TextDecoder();
  let received = '';
  for await (const chunk of stream) {
    received += decoder.decode(chunk, { stream: true });
    if (received.includes(`data: ${expected}`)) return;
  }
  throw new Error('The SSE stream closed before the message arrived.');
}

describe('realtime example', () => {
  const replicas: Replica[] = [];
  const streaming = new AbortController();

  before(async () => {
    // Recorded as soon as each process starts, so `after` stops it even if
    // a later replica never becomes ready.
    for (const port of [await freePort(), await freePort()]) {
      replicas.push(startReplica(port));
    }
    for (const { baseUrl } of replicas) {
      // Any answer proves the socket is bound; this path is deliberately unrouted.
      await waitForHttp(`${baseUrl}/__ready`, { label: `Replica at ${baseUrl}` });
    }
  });

  after(async () => {
    streaming.abort();
    const exitCodes = await Promise.all(replicas.map(({ host }) => stopHost(host)));
    assert.deepEqual(exitCodes, [0, 0], 'a replica did not shut down cleanly');
  });

  it('delivers a message published on replica A to a subscriber on replica B', async () => {
    const [publisher, subscriber] = replicas as [Replica, Replica];
    const events = await fetch(`${subscriber.baseUrl}/events`, { signal: streaming.signal });
    assert.ok(events.ok && events.body !== null, 'replica B did not open an SSE stream');
    const delivered = readUntil(events.body, 'from-replica-a');

    const published = await fetch(`${publisher.baseUrl}/publish`, {
      method: 'POST',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ message: 'from-replica-a' }),
    });
    assert.equal(published.status, 204);

    await Promise.race([
      delivered,
      new Promise<never>((_, reject) => {
        setTimeout(() => reject(new Error('Timed out waiting for replica B.')), 5_000).unref();
      }),
    ]);
  });
});