` writes a starting file. Without a `params` schema, give the params type as a type argument: `defineChannel<{ room: string }>({ ... })`.
## Validate params and payloads
On the server, `params` runs first, and its output is what the subscription is filed under, so a schema that trims merges `' prj_1 '` and `'prj_1'` into one topic. On the client, an event with a `schema` is validated before `effects` runs; a payload that fails is not applied and the error handler gets a `VALIDATION` error:
```ts
// app/lib/projects/project-errors.ts
import { channel } from '@kitr/realtime'
import { projectChannel } from '#app/channels/projects'
import { realtime } from '#app/realtime'
channel(realtime(), projectChannel)
.subscription()
.subscribe(
{ id: 'prj_1' },
{
next: (event) => console.log(event.event, event.data),
error: (error) => console.warn(error),
// => { code: 'VALIDATION', message: 'Invalid payload for projects.updated', status: 422, fieldErrors: { name: [...] } }
},
)
```
Events the channel does not declare are dropped on the client without an error.
> [!WARNING] Emit is not validated
> The server does not validate what you pass to `emit`; schemas only protect clients. Emit data you built from your own records.
## Update the cache from events
`effects(data, params)` returns the same cache effect a [query](/packages/query) mutation returns: `set` writes an entry, `invalidate` marks entries stale so mounted queries refetch, `remove` drops them.
```ts
// app/channels/projects.ts
import { defineChannel } from '@kitr/realtime'
import { z } from 'zod'
const projectSchema = z.object({ id: z.string(), name: z.string(), teamId: z.string() })
export const projectChannel = defineChannel({
name: 'projects',
params: z.object({ id: z.string() }),
events: {
// Write the new value: no request.
updated: {
schema: projectSchema,
effects: (project, { id }) => ({
set: [{ key: ['projects', 'detail', id], data: project }],
invalidate: [['projects', 'list']],
}),
},
// Drop the detail and refetch every list.
deleted: {
effects: (_data, { id }) => ({
remove: [['projects', 'detail', id]],
invalidate: [['projects', 'list']],
}),
},
},
})
```
Use `set` when the event carries the full record. Use `invalidate` when it only says something changed; the query refetches through its normal `fetch`, so permission checks stay on the server. Declare cache writes in `effects`, not with `setQueryData` in a component.
## Emit from the server
`realtime.use(channel, params)` returns an emitter for one topic. Read the server from the container: `ctx.app` in a controller, or the server's app in an action or a job:
```ts
// app/actions/projects/rename.ts
import { defineMutation } from '@kitr/actions'
import { z } from 'zod'
import { projectChannel } from '#app/channels/projects'
import { server } from '#server/main'
export const renameProject = defineMutation('projects.rename', {
input: z.object({ id: z.string(), name: z.string().min(2) }),
handler: async ({ input }) => {
const project = { id: input.id, name: input.name, teamId: 'team_acme' }
// ... save it with your database
server.app.container.use('realtime').use(projectChannel, { id: input.id }).emit('updated', project)
return project
},
})
```
The emitter also reports how many connections are listening, to skip work nobody will see:
```ts
// server/jobs/project-stats.ts
import { projectChannel } from '#app/channels/projects'
import { server } from '#server/main'
const realtime = server.app.container.use('realtime')
const topic = realtime.use(projectChannel, { id: 'prj_1' })
console.log(topic.size)
// => 2 (Ada and Linus have the board open)
console.log(realtime.connections)
// => 3 (open WebSocket connections, subscribed or not)
```
Emit with the params as the server normalised them: with `z.string().trim()`, `{ id: 'prj_1' }` reaches a client that subscribed with `' prj_1 '`, but `{ id: ' prj_1 ' }` reaches nobody. Key order does not matter.
## Subscribe in a component
`useChannel(transport, channel, params, options?)` from `@kitr/realtime/react` subscribes on mount, applies each event's `effects` to the query cache, and unsubscribes on unmount. It needs the [query](/packages/query) provider, which every kit with React includes.
```tsx
// app/components/projects/project-board.tsx
import { useChannel } from '@kitr/realtime/react'
import { projectChannel } from '#app/channels/projects'
import { realtime } from '#app/realtime'
export function ProjectBoard({ id }: { id: string }) {
const live = useChannel(realtime(), projectChannel, { id }, {
onEvent: (event) => {
if (event.event === 'deleted') window.location.assign('/projects')
},
onError: (error) => console.warn(error.code, error.message),
})
return {live.status === 'open' ? 'Live' : live.status}
}
```
`useChannel` returns `{ status, lastEvent, error }`:
| Field | Description |
| --- | --- |
| `status` | `'idle'` before subscribing or with `enabled: false`, `'open'` once subscribed, `'error'` after an error. |
| `lastEvent` | The last `{ event, data }` received, typed as a union of the channel's events. |
| `error` | The last error, a `KitrErrorShape` such as `{ code: 'FORBIDDEN', ... }`. |
| Option | Default | Description |
| --- | --- | --- |
| `enabled` | `true` | `false` skips the subscription, for example until a project id is known. |
| `onEvent(event)` | none | Runs for every valid event, after `effects`. |
| `onError(error)` | none | Runs for subscription errors: params, authorization, payload validation. |
Changing `params` resubscribes.
> [!WARNING] Browser only
> `app/realtime.ts` opens the connection the first time `realtime()` runs, so render subscribing components in the browser only: a route with `ssrRoute(false)` from [seo](/packages/seo), or a component that mounts after hydration.
## Subscribe with useSubscription
`channel(transport, definition)` binds a channel to a transport. Its `subscription()` is a subscription contract that `useSubscription` and `subscriptionFrom` from [query](/packages/query) take directly:
```ts
// app/queries/projects.ts
import { subscriptionFrom } from '@kitr/query'
import { channel } from '@kitr/realtime'
import { projectChannel } from '#app/channels/projects'
import { realtime } from '#app/realtime'
/** Call inside a component, so the socket opens in the browser. */
export const projectSubscription = () =>
subscriptionFrom(channel(realtime(), projectChannel).subscription())
```
```tsx
// app/components/projects/project-live.tsx
import { useSubscription } from '@kitr/query'
import { useMemo } from 'react'
import { projectSubscription } from '#app/queries/projects'
export function ProjectLive({ id }: { id: string }) {
const subscription = useMemo(() => projectSubscription(), [])
const live = useSubscription(subscription, { id })
return {live.status}
}
```
## Show the connection status
`useChannel` reports the subscription. To show whether the socket itself is connected, use `useRealtimeStatus(transport)`:
```tsx
// app/components/connection-badge.tsx
import { useRealtimeStatus } from '@kitr/realtime/react'
import { realtime } from '#app/realtime'
export function ConnectionBadge() {
const status = useRealtimeStatus(realtime())
// => 'connecting' | 'open' | 'closed'
return {status === 'open' ? 'Online' : 'Reconnecting'}
}
```
The WebSocket client reconnects with backoff and resubscribes.
> [!WARNING] No replay
> Events emitted while a client was offline are not replayed. On reconnect, invalidate what matters, or let `refetchOnWindowFocus` catch up.
## Send events from the client
`channel(...).send(params, event, data)` sends an event up a channel. The server runs the channel's `params` schema and `authorize`, then calls its `onMessage`. Here, typing in a comment box tells everyone else on the board. Declare the event so clients can receive it:
```ts
// app/channels/projects.ts
import { defineChannel } from '@kitr/realtime'
import { z } from 'zod'
export const projectChannel = defineChannel({
name: 'projects',
params: z.object({ id: z.string() }),
events: {
typing: { schema: z.object({ user: z.string() }) },
},
})
```
The server adds `onMessage` to its copy of the channel (channels match by `name`), so the handler never reaches the browser:
```ts
// server/routes.ts
r.ws('/ws', {
channels: [
{
...projectChannel,
onMessage: (event, data, params, _ctx, realtime) => {
if (event !== 'typing' || typeof (data as { user?: unknown })?.user !== 'string') return
realtime.use(projectChannel, params).emit('typing', data as { user: string })
},
},
],
})
```
```tsx
// app/components/projects/comment-box.tsx
import { channel } from '@kitr/realtime'
import { projectChannel } from '#app/channels/projects'
import { realtime } from '#app/realtime'
export function CommentBox({ projectId }: { projectId: string }) {
const project = channel(realtime(), projectChannel)
return