Channels

Channels are typed application event streams built on the same realtime runtime as live queries. Define each channel once in channels/; codegen adds it to request contexts as channels, to the generated AppConfig, and to client.channels. The application API stays the same whether delivery uses the zero-infrastructure SSE path or the managed Pusher/Soketi path.

Use channels for transient events such as chat notifications, progress, typing, or presence. The bounded replay ledger is delivery infrastructure, not queryable application history. Persist messages that users must retrieve later in a collection.

Define a channel

import { channel } from "questpie/channels";
import { z } from "zod";

export default channel("chat-room-[roomId]")
	.events({
		message: z.object({ id: z.string(), text: z.string() }),
		typing: z.object({ active: z.boolean() }),
	})
	.authorize({
		subscribe: async ({ params, session, collections }) => {
			if (!session?.user) return false;
			return Boolean(
				await collections.rooms.findOne({
					where: { id: params.roomId, members: { contains: session.user.id } },
				}),
			);
		},
		publish: async ({ params, session, collections }) => {
			if (!session?.user) return false;
			return Boolean(
				await collections.rooms.findOne({
					where: { id: params.roomId, members: { contains: session.user.id } },
				}),
			);
		},
	})
	.presence(({ params, session }) => ({
		id: session!.user.id,
		roomId: params.roomId,
		name: session!.user.name,
	}));

The registry/API key is chatRoom, derived from the default-export filename. The explicit builder string is the stable wire pattern. Renaming the file changes the typed API and causes compile errors at call sites; it does not silently rename the wire channel. [roomId] becomes a required { roomId: string } parameter everywhere.

Run codegen after adding or renaming a channel:

bunx questpie generate

Visibility and authorization

Authorization receives the generated AppContext plus typed params. Use session, collections, db, and services directly. A false result, thrown error, or authorization timeout denies the operation. Server/system contexts may publish without a client publish grant; browser publishes still go through the framework route, authorization, rate limits, and Zod parsing.

Publish on the server

Framework handlers receive a generated, request-bound channels service:

import { route } from "questpie/services";
import { z } from "zod";

export default route()
	.post()
	.schema(z.object({ roomId: z.string(), id: z.string(), text: z.string() }))
	.handler(async ({ input, channels }) => {
		return channels.publish("chatRoom", {
			params: { roomId: input.roomId },
			event: "message",
			data: { id: input.id, text: input.text },
		});
	});

The event name and input are inferred from the channel's Zod schemas. Zod transforms run before the event enters the ledger. publish() resolves to { eventId }, the channel-local ordered id clients use for replay and deduplication.

Publish from collection or global hooks

Use the hook's injected { channels } argument. It carries the same generated types and the mutation-bound database context:

.hooks({
  afterChange: [async ({ data, channels }) => {
    await channels.publish("chatRoom", {
      params: { roomId: data.room },
      event: "message",
      data: { id: data.id, text: data.text },
    });
  }],
})

Do not import the generated app into a collection/hook file, and do not resolve channels later through ambient getContext(). Contextless update/delete calls do not guarantee an ambient scope; the injected hook argument is the reliable and transaction-aware path.

Revoke delivery authority after a domain change

When a membership or authorization write removes access, call the request-bound service in the same transaction:

await channels.revokeAuthority("chatRoom", {
	params: { roomId },
	subject: { kind: "user", id: removedUserId },
	idempotencyKey: `chat-room:${roomId}:${removedUserId}:membership-v2`,
});

The command advances a durable generation for the exact resolved channel/subject and returns { generation, scope }. With SSE, scope: "exact-subscription" closes only that logical binding with access_revoked; other channel bindings on the multiplexed stream remain active. Fresh request context, subscribe authorization, and the presence resolver are evaluated outside database locks, then a short generation check publishes the binding, latest freshly resolved presence payload, and optional presence lease atomically. A stale result and stale member payload publish nothing. Dropped authority notices heal through a demand-driven ledger poll that also stops when internal revocation closes the last local binding.

Pusher/Soketi reports scope: "principal-connections" because its termination API is user-wide. QUESTPIE signs the browser into Pusher with an opaque user id, terminates every current connection for that user, and lets reconnect obtain fresh user and per-channel authorization. Channel authorization waits for signed-user completion and rejects stale sockets or owners. An unrelated still-authorized binding can reconnect; the removed binding cannot. Provider channel auth is explicitly side-effect-free signed-blob encoding; a concurrent authority cut discards the blob before it can reach the browser.

Subscribe and publish on the client

createClient<AppConfig>() exposes one handle per generated channel:

const stop = client.channels.chatRoom.subscribe(
	{ roomId },
	(message) => {
		if (message.event === "message") {
			console.log(message.eventId, message.data.text);
		} else {
			console.log(message.data.active);
		}
	},
	{
		onReady: () => setMutationEnabled(true),
		onNotReady: () => setMutationEnabled(false),
		onError: console.error,
	},
);

const receipt = await client.channels.chatRoom.publish({
	params: { roomId },
	event: "typing",
	data: { active: true },
});

stop();

onReady starts one admitted subscription epoch after authorization and replay catch-up complete. If that epoch ends, onNotReady runs exactly once for the same logical subscriber. A successful reconnect completes fresh authorization and catch-up, then calls onReady again for the next epoch. A subscriber that never reached onReady does not receive onNotReady, and explicitly calling stop() or aborting the subscription removes the callbacks without reporting an epoch end.

Ordinary reconnects report the readiness transition through onNotReady, not onError. A terminal failure after admission calls onNotReady before the existing onError callback. Exceptions thrown by any lifecycle callback are isolated from sibling subscribers, transport cleanup, and later reconnects.

Client publish is server-mediated even on Pusher/Soketi: the request uses the same dynamic auth headers as other client calls, re-evaluates publish authorization, validates the event with Zod, enforces payload/origin/rate limits, and only then appends the ordered event.

For async iteration:

for await (const message of client.channels.chatRoom.iter(
	{ roomId },
	{ signal: controller.signal },
)) {
	consume(message);
}

Call client.channels.destroy() only when disposing the whole client-side channel runtime. Normal components should call the subscription's stop() function or abort their iterator.

Presence

Only a channel with .presence() has a callable presence API:

const members = await client.channels.chatRoom.presence({ roomId });
// readonly Array<{ id: string; roomId: string; name: string }>

Use subscribePresence() for a live typed roster, or presenceIter() when an AbortSignal fits the consumer better:

const stop = client.channels.chatRoom.subscribePresence({ roomId }, (members) =>
	renderRoster(members),
);

for await (const members of client.channels.chatRoom.presenceIter(
	{ roomId },
	{ signal },
)) {
	renderRoster(members);
}

Pusher/Soketi uses native provider presence. SSE uses Postgres leases shared by every app instance and aggregates multiple connections into one member per authenticated principal. Graceful leave is immediate; an ungraceful disconnect converges after the lease expires (30s by default).

TanStack Query

The query-options proxy exposes channel subscriptions without adding a separate React provider or bespoke hook:

const { data: messages = [] } = useQuery(
	q.channels.chatRoom.subscription({ roomId }),
);

const { data: members = [] } = useQuery(
	q.channels.chatRoom.presence({ roomId }),
);

The subscription query starts with [] and appends each typed { event, eventId, data } message. The presence query keeps only the latest roster snapshot. Both abort their underlying iterator on unmount.

Transport selection

SSE is the default edge transport. Channel subscribe uses the framework SSE/control routes, publish uses HTTP POST, and presence uses the application Postgres database as a cross-instance lease register. No client configuration changes are needed.

To use managed WebSockets and native presence, install the optional Pusher peers and select the preset under realtime:

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

const managed = pusherRealtime({
	appId: env.PUSHER_APP_ID,
	key: env.PUSHER_KEY,
	secret: env.PUSHER_SECRET,
	cluster: env.PUSHER_CLUSTER,
	// For Soketi, provide host/wsHost/ports instead.
});

export default runtimeConfig({
	realtime: { ...managed },
});

The client discovers the selected transport from /channels/config; client.channels.* call sites do not change. After a managed-provider reconnect, the established subscription reauthorizes a bounded /channels/replay drain from its last applied event id. Live provider events arriving during that drain are buffered behind the replay cursor, then deduplicated and released in order.

Pusher signed-user authentication uses the current dynamic auth headers and cookies. If login or logout changes identity without replacing the browser socket, call both client.realtime.destroy() and client.channels.destroy() (or recreate the client) before subscribing again so the shared provider connection is recreated under the new identity.

Delivery and security contracts

• Framework events are ordered per resolved channel and are never coalesced. Reconnect may replay an event id; clients deduplicate it.
• Falling behind the bounded replay horizon produces an explicit channel gap rather than invented state. Recover from a persisted collection or subscribe from now.
• Server delivery queues, managed-provider replay races, and client async iterators are bounded by event count and serialized UTF-8 bytes. A slow consumer terminates with an explicit error; ordered events are not silently dropped.
• A managed-provider edge sends an authenticated control probe every 15 seconds. Missing three probes expires the independent client lease and releases admission, query, and CRDT bindings even when the provider cannot report an abrupt browser disappearance.
• Payloads must be JSON-serializable. The complete canonical { eventId, event, data } envelope must fit 10,000 UTF-8 bytes, so the application data budget is slightly smaller and varies with the event id and event name. Typed datetime metadata is part of those measured bytes.
• Ordered events, replay, and presence preserve nested Date values through SSE and Pusher/Soketi without heuristic ISO-string revival. The compatible versioned wire format and reserved metadata key are specified in Temporal values.
• Cookie-authenticated authority routes require an exact trusted Origin. Additional origins belong in realtime.channelSecurity.trustedOrigins.
• Client publishes use independent per-session and per-principal token buckets (defaults: 10/s, burst 20).

Related

• Reactive Apps & Performance, channel sizing, bounded event state, React isolation, and live presence lifecycle.
• Realtime, live collection/global snapshots.
• Realtime adapter, SSE versus Pusher/Soketi selection, reconciliation, admission, and privacy.
• TanStack Query, the complete option-builder surface.