Copy page

SDK Library

SSE Standard

Primebrick uses Server-Sent Events (SSE) to push real-time updates from the BE to connected FE clients. SSE replaces polling for data that changes event-driven (e.g. service health status).

The SDK provides three building blocks: createSseWriter (wire format), createSseEventBus (in-process distribution), and bridgeNatsToSse (NATS → bus bridge). SSE endpoints exist only on the BE — microservices do NOT expose SSE. The BE is the sole public attack surface.

Architecture

Code
// NATS is the fanout mechanism for multi-instance BE deployments.
// Each BE instance subscribes to the same NATS subjects and bridges
// them to its local SseEventBus. SSE endpoints subscribe to the bus
// and forward events to connected clients.

// Microservice → NATS → BE (all instances) → local SseEventBus → SSE clients

Multi-instance fanout is handled by NATS, not Redis pubsub. Redis is used only for caching. Every BE instance subscribes to the same NATS subjects; each bridges events to its own local SseEventBus, which feeds its own SSE connections.

SDK building blocks

createSseWriter

Sets SSE headers on an Express Response and returns an SseWriter that handles the W3C EventSource wire format. Uses extJsonStringify for BigInt-safe serialization of event data.

Code
import { createSseWriter } from "@primebrick/sdk";
import type { Response } from "express";

function sseEndpoint(res: Response) {
  const writer = createSseWriter(res);

  // Send an event
  writer.send({
    id: "service:heartbeat:1700000000:abc",
    event: "service.heartbeat",
    data: { code: "emailsender", status: "online" },
  });

  // Keep-alive comment (no event type, just a comment line)
  writer.comment("keep-alive");

  // Close on disconnect
  res.on("close", () => writer.close());
}

createSseEventBus

In-process event bus. Typically created once as a singleton per BE process and shared across all SSE endpoints that need the same event stream.

Code
import { createSseEventBus } from "@primebrick/sdk";

export const serviceEventsBus = createSseEventBus();

// Emit from anywhere in the BE process:
serviceEventsBus.emit({
  id: "service:heartbeat:1700000000",
  event: "service.heartbeat",
  data: { code: "emailsender", status: "online" },
});

// Subscribe in an SSE endpoint:
const sub = serviceEventsBus.subscribe((event) => {
  writer.send(event);
});
// Cleanup on disconnect:
sub.unsubscribe();

bridgeNatsToSse

Subscribes to NATS subjects and forwards each message as an SseEvent on the bus. Returns a cleanup function that unsubscribes all NATS subscriptions.

Code
import { bridgeNatsToSse, NatsClient, SERVICE_SUBJECTS } from "@primebrick/sdk";

const cleanup = await bridgeNatsToSse(NatsClient, serviceEventsBus, [
  {
    subject: SERVICE_SUBJECTS.HEARTBEAT,
    eventType: "service.heartbeat",
    transform: (p) => ({
      id: `hb:${p.code}:${Date.now()}`,
      data: p,
    }),
  },
  {
    subject: SERVICE_SUBJECTS.REGISTER,
    eventType: "service.register",
    transform: (p) => ({
      id: `reg:${p.code}:${Date.now()}`,
      data: p,
    }),
  },
]);

// On graceful shutdown:
cleanup();

SSE development standard

When creating a new SSE endpoint in Primebrick, follow these rules:

1. URL convention

Code
GET /api/v1/<module>/<resource>/events

Example: GET /api/v1/system/services/events

Each SSE endpoint is specific to a resource. There is no generic event hub.

2. Auth and RBAC

Reuse the same middleware as regular REST endpoints: authMiddleware() + rbacHandler([Permission.X]). Cookie-based auth (credentials: 'include') works with SSE — cookies are sent automatically.

If the token expires during an SSE connection, the BE sends an error event with { type: "auth_expired" } and closes the connection. The FE refreshes the token and reconnects.

3. Response headers

createSseWriter sets these headers automatically:

Header	Value	Purpose
Content-Type	text/event-stream; charset=utf-8	SSE wire format
Cache-Control	no-cache, no-transform	Prevent proxy buffering
Connection	keep-alive	Persistent connection
X-Accel-Buffering	no	Disable nginx buffering
X-Content-Type-Options	nosniff	Security header

4. Event format

Every event follows the W3C EventSource format:

Code
id: <event-id>
event: <event-type>
data: <json-single-line>

• id: Unique, deterministic where possible. Used for Last-Event-ID on reconnect and FE-side deduplication.
• event: Type in dot notation (e.g. service.heartbeat, snapshot).
• data: Single-line JSON serialized via extJsonStringify. BigInt values are preserved; Date objects become ISO strings.

5. Snapshot on connect

Immediately after the SSE connection is established, send a snapshot event with the current state. This gives the FE the full picture without needing a separate REST call.

Code
writer.send({
  id: `snapshot:${Date.now()}`,
  event: "snapshot",
  data: { services: await repo.findAll() },
});

6. Keep-alive

Send a comment every 15 seconds to keep the TCP connection alive and detect disconnected clients (write fails on broken sockets):

Code
const keepAlive = setInterval(() => writer.comment("keep-alive"), 15_000);

7. Connection lifecycle

1. Connect: auth middleware validates → createSseWriter sets headers → send snapshot → subscribe to SseEventBus
2. Stream: forward bus events to client + keep-alive every 15s
3. Disconnect (req.on('close')): unsubscribe from bus, clear keep-alive interval, close writer
4. Graceful shutdown (SIGTERM): emit error event with { type: "server_shutdown" }, close all connections, close bus

8. Server timeout

Set server.timeout = 300_000 (5 minutes) on the HTTP server. This is a socket inactivity timeout — it resets on every res.write(), so active SSE connections with 15s keep-alive never hit it. It only fires when no data flows for 5 minutes (zombie connections), allowing the server to reclaim resources.

Never set server.timeout = 0. Zero timeout means connections never expire, leading to zombie accumulation, memory leaks, and GC pressure over time. 5 minutes is the trade-off: survives micro-oscillations (lost 1-2 keep-alives) but reclaims dead connections.

9. FE client

The FE uses @microsoft/fetch-event-source (not native EventSource) for control over reconnection, backoff, and headers. The FE client implements:

• Exponential backoff: 1s → 2s → 4s → 8s → 16s → 30s (cap)
• Visibility-aware: pauses when tab is hidden, resumes when visible
• Cookie auth: credentials: 'include' (same as REST)

10. FE deserialization

The FE MUST use extJsonParse() (from $lib/api-ext) to parse SSE event data — never native JSON.parse(). This preserves BigInt values across the wire. Using JSON.parse() would silently lose precision on integers larger than Number.MAX_SAFE_INTEGER.

11. NATS as fanout, not Redis

NATS is the messaging backbone for SSE fanout across BE instances. Redis pubsub is NOT used for SSE — Redis is cache-only in Primebrick. Every BE instance subscribes to the same NATS subjects and bridges them to its local SseEventBus.

12. Microservices do NOT expose SSE

SSE endpoints exist only on the BE. Microservices communicate via NATS (register, heartbeat, unregister). The BE bridges NATS events to SSE clients. This keeps the BE as the sole public attack surface.

Complete endpoint example

Code
import { Router } from "express";
import { createSseWriter, type SseEventBus, Permission } from "@primebrick/sdk";
import { rbacHandler } from "../../http/rbac-handler.js";
import { asyncHandler } from "../../http/async-handler.js";
import { ServiceRegistryRepo } from "../proxy/service-registry-repo.js";
import { getPool } from "../../db/pool.js";
import { serviceEventsBus } from "../proxy/service-events-bus.js";

const KEEPALIVE_MS = 15_000;

export function servicesEventsRouter(): Router {
  const router = Router();

  router.get(
    "/api/v1/system/services/events",
    rbacHandler([Permission.AUTHENTICATED_USER]),
    asyncHandler(async (req, res) => {
      const writer = createSseWriter(res);

      // 1. Send initial snapshot
      const repo = new ServiceRegistryRepo(getPool());
      const services = await repo.findAll();
      writer.send({
        id: `snapshot:${Date.now()}`,
        event: "snapshot",
        data: { services },
      });

      // 2. Subscribe to event bus
      const sub = serviceEventsBus.subscribe((event) => writer.send(event));

      // 3. Keep-alive
      const ka = setInterval(() => writer.comment("keep-alive"), KEEPALIVE_MS);

      // 4. Cleanup on disconnect
      req.on("close", () => {
        sub.unsubscribe();
        clearInterval(ka);
        writer.close();
      });
    }),
  );

  return router;
}

Next steps

• NATS Client — the messaging backbone for SSE fanout
• Service Registration — lifecycle events that feed SSE
• Ext-JSON — BigInt-safe serialization used by the SSE writer
• API Reference — full type definitions for SSE modules