Realtime

Two seams sit under that, and you can swap either one. A change broker carries wake-up notices between your app instances. A client transport carries authorized frames out to browsers. Application code calls live() and client.channels.* and never learns which pair you picked.

What you get with no configuration

Nothing. Write no realtime block and realtime still runs, because change capture, the POST /realtime SSE endpoint and the channel routes ship in the core module. On a db: { url } app you get:

realtime: true is accepted and resolves to {}, which is exactly what omitting the key already gives you. Reach for the object form when you want to change a default.

The two primitives

Live queries carry a fresh snapshot of a collection or global, so use them for lists, detail views and counters. Channels carry a typed application event, so use them for progress, typing and notifications.

const unsubscribe = client.collections.posts.live(
	{ where: { status: "published" }, orderBy: { createdAt: "desc" } },
	(snapshot) => render(snapshot.docs),
);

The callback fires once with the current snapshot, then again on each relevant change. liveIter() yields the same snapshots as an async generator. Client realtime has the lifecycle options and the TanStack Query integration. Channels covers channel(), authorization, presence and replay.

The adapters

Each seam has its own choices and you may mix them. Redis for the notice with SSE at the edge is a normal combination.

pusherRealtime() returns { changeBroker, clientTransport }, so it fills both seams at once. The isolated entry point is the only thing that loads those two optional peers.

Swapping the broker

Build the client, hand it over, change nothing else:

import { runtimeConfig } from "questpie/app";
import { redisStreamsChangeBroker } from "questpie/adapters/redis-streams";
import { createClient } from "redis";

import env from "./env";

const redis = createClient({ url: env.REDIS_URL });
await redis.connect();

export default runtimeConfig({
	db: { url: env.DATABASE_URL },
	realtime: {
		changeBroker: redisStreamsChangeBroker({
			client: redis,
			// stream: "questpie:realtime:v2", // default
			// blockMs: 5000, batchSize: 100, maxLen: 10_000,
		}),
	},
});

Every instance tails the stream from $ with its own XREAD cursor, so each one receives every wake rather than competing for it. XADD applies MAXLEN ~ 10000. Only xAdd and xRead are mandatory on the client, and a node-redis client is duplicated automatically for the blocking read. Pass reader when your client cannot duplicate itself.

Moving to managed WebSockets is the same edit against the other seam:

import { pusherRealtime } from "questpie/adapters/pusher";

export default runtimeConfig({
	db: { url: env.DATABASE_URL },
	realtime: pusherRealtime({
		appId: env.PUSHER_APP_ID,
		key: env.PUSHER_KEY,
		secret: env.PUSHER_SECRET,
		cluster: env.PUSHER_CLUSTER,
		// Soketi: host, port, wsHost, wsPort, wssPort, useTLS
	}),
});

live() and client.channels.* are untouched by either edit. The client learns which transport it is on from GET /realtime/config.

Configuring the pg broker by hand

You rarely need to. Configure it explicitly for a non-default channel, to reuse an existing pg client, or to aim realtime at a second database.

import { pgNotifyChangeBroker } from "questpie/adapters/pg-notify";

realtime: {
	changeBroker: pgNotifyChangeBroker({ connectionString: env.REALTIME_DB_URL }),
}

Give it one connection source. With none of client, connection or connectionString it throws on start, and if you pass several it does not complain, it just prefers client, then connection. A client you hand over is left open on stop(), one built from a connection string is closed. Channel names are checked against /^[a-zA-Z0-9_]+$/ at construction.

Turning it down

rowLiveQueries: false is the app-wide switch for collection and global row topics, and leaves channels, CRDT and outbox capture running. Per collection, .options({ realtime: false }) opts one table out. Both refuse the topic at admission with REALTIME_TOPIC_REJECTED rather than degrading quietly.

nativeDeltas is false by default. Until you turn it on, a topic that qualifies for keyed deltas is served as a snapshot instead, so correctness never waits on a rollout. Scalable realtime is the page for high-fanout modeling.

Writing your own broker

Implement three methods and wire the result as realtime.changeBroker.

interface ChangeBroker {
	start(input: {
		onWake: (wake: ChangeWake) => void;
		onError: (error: unknown) => void;
		onStateChange?: (state: ChangeBrokerState) => void;
	}): Promise<void>;
	publish(wake: ChangeWake): Promise<void>;
	stop(): Promise<void>;
}

Wakes may be lost, duplicated, delayed, reordered or coalesced, and they must never carry rows, snapshots or credentials. Reconciliation against the durable outbox is what supplies correctness, which is why a lossy broker is a safe thing to write. ChangeBroker, ClientTransport, RealtimeConfig and the outbox event types are exported from questpie/realtime.

Where each topic lives