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.

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);
		}
	},
	{ onError: console.error },
);

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

stop();

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.

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.
• A slow SSE consumer is disconnected after bounded event/byte queues fill; ordered events are not silently dropped.
• Payloads must be JSON-serializable and at most 10,000 UTF-8 bytes.
• 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.