@apco/realtime

Server-Sent Events (SSE) realtime for APCO apps. There is no platform broker — the realtime hub runs inside your SSR app's process, so it works anywhere the app runs. Requires Node.js ≥ 22 (ESM only); React ≥ 18 is an optional peer dependency (only for the hook). See the realtime guide for a walkthrough.

npm install @apco/realtime

Entry point	Exports
@apco/realtime	createRealtime, DEFAULT_MAX_CLIENTS, HARD_MAX_CLIENTS, types Realtime, RealtimeOptions, RealtimeEvent, PublishResult, RealtimeAdapter
@apco/realtime/client	subscribe, types SubscribeOptions, Subscription, RealtimeClientEvent
@apco/realtime/events	defineRealtimeEvents, eventsFor, type RealtimeEventDefinitions
@apco/realtime/react	useRealtime, types UseRealtimeOptions, RealtimeClientEvent, Subscription
@apco/realtime/postgres	createPostgresAdapter, types PostgresAdapterOptions, PostgresAdapterError, SqlLike

Server (@apco/realtime)

createRealtime()

import { createRealtime } from "@apco/realtime";

const rt = createRealtime({ channels: ["todos"], maxClients: 100 });

// Publish from any route:
rt.publish("todos", { event: "todo:created", data: row });

// Mount the SSE endpoint (Fetch-API server, e.g. Hono):
app.get("/api/events", (c) => rt.handler(c.req.raw));

Signatures:

interface RealtimeOptions {
  /** Allowed channel names. Omit to allow any channel. */
  channels?: string[];
  /** Heartbeat interval in ms. Default: 30_000. */
  heartbeatMs?: number;
  /** Per-channel throttle. Pass false to disable. */
  throttle?: false | {
    maxTokens?: number;
    refillIntervalMs?: number;
    refillTokens?: number;
  };
  /** Called when a publish is dropped by channel filtering or throttling. */
  onDrop?: (drop: { channel: string; reason: "channel_not_allowed" | "throttled"; event?: string }) => void;
  /**
   * Max concurrent SSE clients. Default: 25 (DEFAULT_MAX_CLIENTS).
   * Clamped to the hard ceiling of 1000 (HARD_MAX_CLIENTS).
   */
  maxClients?: number;
  /** Cross-instance transport. Omit to stay strictly in-process (default). */
  adapter?: RealtimeAdapter;
  /** Called when the adapter rejects a publish or fails to subscribe. */
  onAdapterError?: (err: unknown, context: { op: "publish" | "subscribe" }) => void;
  /**
   * SSE resume buffer. Default `{ bufferSize: 100 }` (clamped to [0, 10_000]).
   * Every delivered event is stamped with a per-instance `id` and retained in a
   * bounded ring buffer; a client reconnecting with a matching Last-Event-ID
   * replays what it missed. Pass `false` (or `bufferSize: 0`) to disable replay
   * AND drop the `id:` line (byte-identical legacy wire format).
   */
  replay?: { bufferSize?: number } | false;
  /**
   * Per-channel authorization hook, run at subscribe time for each requested
   * channel. Return false to filter a channel out; configuring it makes
   * `handler()` async. See "Channel authorization" below.
   */
  authorize?: (channel: string, request: Request | IncomingMessage) => boolean | Promise<boolean>;
}

interface RealtimeEvent {
  channel: string;      // logical channel
  event?: string;       // SSE event: type; defaults to "message"
  data: unknown;        // JSON-serializable payload
  id?: string;          // SSE id: — stamped per-instance when replay is on (see below)
}

interface Realtime {
  publish(channel: string, event: Omit<RealtimeEvent, "channel">): PublishResult;
  handler(request: Request): Response | Promise<Response>;  // Fetch-API SSE stream (async only when `authorize` is set)
  node(req: IncomingMessage, res: ServerResponse): void;   // Node http adapter
  clientCount(): number;                                    // currently connected clients
  close(): Promise<void>;                                   // unsubscribe adapter (no-op without one)
}

type PublishResult =
  | { delivered: true }
  | { delivered: false; reason: "channel_not_allowed" | "throttled" };

function createRealtime(opts?: RealtimeOptions): Realtime;

const DEFAULT_MAX_CLIENTS = 25;
const HARD_MAX_CLIENTS = 1000;

publish(channel, event)

Publishes are still safe to fire-and-forget, but now return a delivery result for observability. Two drop rules to know about:

1. Channel allow-list — when channels was passed to createRealtime, publishes to any other channel return { delivered: false, reason: "channel_not_allowed" }.
2. Per-channel throttle — publishes are rate-limited per channel by a token bucket (burst 10, refilling ~10 events/second by default). Events beyond that return { delivered: false, reason: "throttled" }, not queued. Coalesce hot paths (e.g. publish one "changed" event and let clients refetch) rather than publishing per row.

Use onDrop for central logging or counters:

const rt = createRealtime({
  channels: ["todos"],
  onDrop: (drop) => console.warn("dropped realtime event", drop.reason),
});

Event definitions

For bundled apps, define channel/event names once and reuse them on server and client:

import { defineRealtimeEvents, eventsFor } from "@apco/realtime/events";

export const realtimeEvents = defineRealtimeEvents({
  todos: ["todo:created", "todo:updated", "todo:deleted"],
});

rt.publish("todos", { event: "todo:created", data: row });

subscribe({
  url: "/api/events",
  channels: ["todos"],
  events: eventsFor(realtimeEvents, "todos"),
  onEvent: () => refetch(),
});

handler(request) — Fetch API

Returns a Response streaming text/event-stream (headers: Cache-Control: no-cache, Connection: keep-alive). Honors an optional ?channels=a,b query param — only events on those channels are delivered to that client. When clientCount() is already at maxClients, responds 503 "Too many clients". A comment heartbeat (: heartbeat) is written every heartbeatMs to keep proxies from idling the connection out.

Without authorize the handler is synchronous (returns a Response) and byte-for-byte compatible with earlier versions when replay is disabled. Configuring authorize makes it return Promise<Response> (it awaits the per-channel checks before building the stream) — await it or return the promise directly; Fetch-API servers like Hono accept both.

node(req, res) — Node http adapter

Same semantics as handler for plain node:http / Express-style servers:

import { createServer } from "node:http";
const server = createServer((req, res) => {
  if (req.url?.startsWith("/api/events")) return rt.node(req, res);
  // ...
});

Wire format and sanitization

Each event is written as a named SSE event whose data: is a JSON envelope. When the replay buffer is enabled (the default), an id: line carrying the per-instance resume id precedes it:

id: 3f9a1c2b4d5e:42
event: todo:created
data: {"channel":"todos","data":{"id":1,"text":"ship docs"}}

• The id: line is the per-instance resume id (<instanceId>:<seq>) used by SSE resume. It is present only while replay is enabled — set replay: false (or bufferSize: 0) to drop it entirely and restore the exact legacy wire bytes (event: + data: only). Like event names, a caller-supplied id is sanitized (CR/LF stripped) before it hits the stream.
• The event: name and the channel are sanitized (CR/LF stripped) before hitting the stream, so user-influenced names cannot inject forged SSE fields; an event name that sanitizes to empty falls back to message.
• data is JSON-stringified, which escapes any remaining control characters — payloads can never break SSE framing.

SSE resume

By default every delivered event — local publishes and adapter-inbound events — is stamped with a per-instance id <instanceId>:<seq> (a random hex instanceId per createRealtime call plus a monotonic counter) and kept in a bounded ring buffer (replay.bufferSize, default 100, clamped to [0, 10_000]). When a client reconnects with a Last-Event-ID header or a ?lastEventId= query param (the param wins — the custom client uses it), the server:

1. parses the resume id and checks the instanceId matches this instance;
2. if so, synchronously replays every buffered event with a greater seq that passes the connection's channel filter (and authorize) before live events flow;
3. otherwise replays nothing.

This is best-effort, per instance. A foreign or garbled id, a cross-instance reconnect (a different replica, or after a restart), or a gap larger than the buffer all fall through to no replay — so clients should still refetch on connect for authoritative state (see the scaling guide). The id: line and buffer are the only additions; the Postgres adapter envelope is unchanged — ids are stamped locally and never cross the adapter.

Setting replay: false (or bufferSize: 0) disables the buffer and the id: line together.

Channel authorization

Pass authorize(channel, request) to gate subscriptions per channel. It runs at subscribe time (in both handler() and node()) for every requested channel and returns boolean | Promise<boolean>:

const rt = createRealtime({
  channels: ["todos"],
  authorize: (channel, request) => {
    const cookie = request.headers.get?.("cookie") ?? "";     // Fetch Request
    return hasValidSession(cookie);                            // your check
  },
});

Semantics:

• Unauthorized channels are silently filtered out of the subscription. If the request named channels and none survive, the subscription is rejected with 403.
• A subscribe-all request (no ?channels=) is rejected with 400 when authorize is set — a wildcard can't be authorized per channel, so clients must name their channels.
• A throwing hook fails closed → 403.
• Configuring authorize makes handler() asynchronous (see above); node() waits for the checks before writing any headers. Publish, throttle, allowlist, and adapter behavior are unchanged.

request is a Fetch Request in handler() and a node:http IncomingMessage in node() — read headers accordingly (request.headers.get("cookie") vs request.headers.cookie).

Capacity planning

Your plan's max_sse_connections limit (maxSseConnections in GET /account/usage) is the number you should pass as maxClients — the library defaults to a conservative 25 and hard-caps at 1000 regardless of what you pass.

Cross-instance adapter (@apco/realtime)

By default the realtime hub is in-process — two instances of the same app (production + apco dev --sync, or multiple replicas) can't see each other's events. Pass an adapter to createRealtime to bridge them. See the scaling guide for the full walkthrough and payload guidance.

interface RealtimeAdapter {
  /** Broadcast a locally published event to other instances. */
  publish(evt: RealtimeEvent): void | Promise<void>;
  /** Receive events from other instances; resolves to an unsubscribe fn. */
  subscribe(onEvent: (evt: RealtimeEvent) => void): Promise<() => Promise<void>>;
  /** Optional cleanup. */
  close?(): Promise<void>;
}

Behavior when an adapter is configured:

• A local publish() that passes the channel allowlist and throttle is delivered locally and forwarded to adapter.publish(). Adapter failures are routed to onAdapterError and never thrown back into the caller — publish() stays synchronous and still returns its PublishResult.
• Events arriving from the adapter are delivered straight to local subscribers. They skip the throttle (already throttled at their origin) and are never re-forwarded to the adapter (no loops), but the channel allowlist still applies (dropped events call onDrop with reason: "channel_not_allowed").
• close() unsubscribes from the adapter and calls adapter.close?.(). It's a no-op when no adapter was configured, idempotent (safe to call twice or concurrently), and resolves even if the adapter's teardown fails (the error is routed to onAdapterError).
• Startup gap: there is a brief window between createRealtime() returning and the adapter's subscription (e.g. Postgres LISTEN) being established — events published by peers during that window are missed, so refetch on connect (see the scaling guide).

createPostgresAdapter() (@apco/realtime/postgres)

A ready-made adapter over Postgres LISTEN/NOTIFY, built on a postgres.js sql instance (as provided by @apco/db):

import { createDb } from "@apco/db";
import { createRealtime } from "@apco/realtime";
import { createPostgresAdapter } from "@apco/realtime/postgres";

const db = createDb();
const rt = createRealtime({
  channels: ["todos"],
  adapter: createPostgresAdapter({ sql: db.sql }),
});

interface PostgresAdapterOptions {
  /** A postgres.js `Sql` (e.g. `createDb().sql`). */
  sql: SqlLike;
  /** Postgres NOTIFY channel. Default: "apco_realtime". */
  channel?: string;
  /** Called on payload/transport errors; never throws into the caller. */
  onError?: (err: PostgresAdapterError) => void;
}

interface PostgresAdapterError {
  code: "payload_too_large" | "notify_failed" | "listen_failed" | "unlisten_failed" | "malformed_message";
  channel: string;
  event?: string;
  cause?: unknown;
}

function createPostgresAdapter(options: PostgresAdapterOptions): RealtimeAdapter;

• One Postgres NOTIFY channel (default apco_realtime) carries a JSON envelope { origin, channel, event, data }. Each adapter instance has a random origin id and drops its own echoes (a NOTIFY is delivered to the notifying session's listener too).
• sql.listen() gets its own dedicated connection courtesy of postgres.js; sql.notify() reuses the pool.
• @apco/realtime never takes a runtime dependency on postgres — the sql param is typed structurally (SqlLike). Pass db.sql or a postgres() client directly.

NOTIFY payload limit (~8 KB)

Postgres caps NOTIFY payloads at ~8000 bytes. Envelopes over a conservative 7500-byte serialized budget are not sent — the adapter reports payload_too_large through onError and the event stays local-only. Publish compact "invalidate" events and refetch state over HTTP instead of shipping large payloads; see the scaling guide.

Client (@apco/realtime/client)

subscribe()

import { subscribe } from "@apco/realtime/client";

const sub = subscribe({
  url: "/api/events", // relative works in the browser; absolute required in Node
  channels: ["todos"],
  events: ["todo:created", "todo:updated", "todo:deleted"],
  onEvent: (evt) => console.log(evt.channel, evt.event, evt.data),
});

// later:
sub.close();

Signatures:

interface SubscribeOptions {
  /** SSE endpoint URL. Relative allowed in a browser; absolute required in Node. */
  url: string;
  /** Channels to subscribe to (sent as the ?channels= query param). */
  channels?: string[];
  /**
   * Named SSE event types to listen for. EventSource only delivers named
   * events to explicit listeners — anything published with a custom `event`
   * must be listed here. Default (unnamed) "message" events are always delivered.
   */
  events?: string[];
  /** Called for every received event. */
  onEvent: (event: RealtimeClientEvent) => void;
  /** Called when EventSource opens. */
  onOpen?: () => void;
  /** Called on status transitions. */
  onStatus?: (status: RealtimeStatus) => void;
  /** Called before a reconnect attempt is scheduled. */
  onReconnect?: (attempt: number, delayMs: number) => void;
  /** Called on connection errors (before a reconnect is scheduled). */
  onError?: (error: Event) => void;
  /** Parse JSON payloads into `event.json` when possible. Default: false. */
  parseJson?: boolean;
  /** Initial reconnect delay in ms. Default: 1000. */
  reconnectMs?: number;
  /** Maximum reconnect delay in ms. Default: 30_000. */
  maxReconnectMs?: number;
}

interface RealtimeClientEvent {
  event: string;               // SSE event type ("message" for unnamed)
  data: string;                // payload as a string (JSON.parse it yourself)
  json?: unknown;              // parsed payload when parseJson is true and parsing succeeds
  channel: string | undefined; // from the server envelope, when present
}

type RealtimeStatus = "connecting" | "open" | "reconnecting" | "closed";

interface Subscription {
  close(): void; // close the connection and stop reconnecting
  status(): RealtimeStatus;
}

function subscribe(opts: SubscribeOptions): Subscription;

Behavior notes:

• Named events must be listed. This is an EventSource platform rule, not a library quirk: an event published with event: "todo:created" is only delivered if "todo:created" is in events. Forgetting this is the most common "no events arriving" bug.
• Reconnects use exponential backoff: reconnectMs * 2^attempt, capped at maxReconnectMs (1 s → 2 s → 4 s → ... → 30 s by default); the attempt counter resets on a successful open. onReconnect receives the attempt number and scheduled delay. close() stops everything.
• Resume on reconnect. The client remembers the last SSE id it received and, on a manual reconnect only, appends it as ?lastEventId= so a server with the replay buffer can replay what was missed. (It has to: recreating the EventSource on reconnect drops the native Last-Event-ID header, so the query param restores it.) Resume is best-effort — cross-instance reconnects and buffer overruns still need an application refetch, so keep a refetch-on-onOpen.
• URLs: relative URLs resolve against location.href in the browser. In Node (no location) a relative URL throws immediately with a descriptive error — pass an absolute URL.
• evt.data is always a string: the client unwraps the server's {channel, data} envelope, re-stringifying non-string payloads. Pass parseJson: true to also receive evt.json when parsing succeeds.

React hook (@apco/realtime/react)

import { useRealtime } from "@apco/realtime/react";

const { connected } = useRealtime({
  url: "/api/events",
  channel: "todos",
  events: ["todo:created", "todo:updated", "todo:deleted"],
  onEvent: (evt) => refetch(),
});

Signature:

interface UseRealtimeOptions {
  url: string;
  channel: string;      // single channel (the hook wraps subscribe with channels: [channel])
  events?: string[];    // named SSE event types
  onEvent: (event: RealtimeClientEvent) => void;
  enabled?: boolean;    // default true; false tears the connection down
}

type RealtimeStatus = "connecting" | "open" | "reconnecting" | "closed";

function useRealtime(opts: UseRealtimeOptions): { connected: boolean; status: RealtimeStatus };

Behavior notes:

• connected flips to true when EventSource opens and to false on reconnecting, closed, errors, and unmount.
• onEvent is kept in a ref — an unstable callback identity does not resubscribe. The subscription is re-created when url, channel, enabled, or the joined events list changes (a new array with the same names does not resubscribe).
• Cleans up (closes the EventSource, stops reconnecting) on unmount and dependency changes.

Example

The examples/realtime-todo app is the end-to-end reference: Hono SSR + @apco/db + @apco/realtime, where API routes write to Postgres and publish todo:* events, and the frontend subscribes and refetches. Scaffold its manifest with apco init --template realtime-todo.