# Realtime

Define a live channel once and push updates to every open screen, over WebSockets, Socket.IO or server events.

A chat, a notification bell, a project board that updates while you look at it: all of them need the server to push to screens that are already open. `kitr/realtime` lets you define a typed **channel** once, emit events on it from the server, and have every subscribed screen update its query cache. It runs over [crossws](https://crossws.unjs.io) WebSockets by default, or [Socket.IO](https://socket.io/docs/v4/) or Server-Sent Events, and the channel code stays the same on all three.

It runs on both sides: the server half (`kitr/realtime/server`) accepts connections and emits; the client half (`kitr/realtime`, `kitr/realtime/react`) subscribes. For one-off server calls that return a value, use [actions](/packages/actions) or [api](/packages/api) instead; realtime is for events the server sends when something changes.

This guide follows Acme's project board at `app.acme.com`. When Grace renames a project, Ada and Linus see the new name without reloading.

## Install

```sh
kitr add realtime
```

What `kitr add` does:

- Adds the [server](/packages/server) package first if the project does not have it yet.
- Installs `@kitr/realtime` and `crossws`. In a React app it also installs `partysocket`, the reconnecting WebSocket client.
- Adds `'@kitr/realtime'` to `packages` in `kitrrc.ts`. Realtime registers no provider.
- Writes `config/realtime.ts` (shown under Configure).
- Writes `app/channels/example.ts`, a sample channel named `example` with one `message` event.
- Writes `server/realtime.ts`, which creates the WebSocket server with `exampleChannel`.
- Adds `realtime.attach(server, '<path>')` to `server/main.ts` when that file exists, with the path from `config/realtime.ts`.
- In a React app, writes `app/realtime.ts`, which opens one client connection per tab on first use.
- Regenerates `config/index.ts`, the config list the standalone server imports.
- Adds a `## kitr: realtime` section to `AGENTS.md` that says where channels live, how the server emits and how components subscribe, and how to switch transports.

`kitr add realtime` asks no questions. It checks for React instead: a project without React (a pure backend) gets the `/ws` path and no client files. Pass `--dry-run` to print the changes without writing them:

```sh
kitr add realtime --dry-run
```

The `fullstack` and `server` kits and the `realtime` kit already include it. `kitr remove realtime` removes the package entry, the `AGENTS.md` section, `config/realtime.ts` and the `crossws`, `partysocket`, `socket.io` and `socket.io-client` dependencies. It leaves `app/channels/`, `server/realtime.ts`, `app/realtime.ts` and the `attach` line in `server/main.ts`, because you have likely edited them.

## Configure

In a React app, `kitr add realtime` writes this file:

```ts
// config/realtime.ts
import { defineConfig } from 'kitr/realtime'

export default defineConfig({
  // websocket (default), socket.io or sse: the channels and hooks stay the same.
  transport: 'websocket',
  path: '/api/ws',
})
```

In a project without React, the path is `'/ws'` instead of `'/api/ws'`.

The client reads this file through `createRealtimeClient(config)`. The server does not read it: `server/realtime.ts` decides which server to create, and `server/main.ts` passes the path to `attach`. Keep the two in step when you change either.

| Option | Default | Description |
| --- | --- | --- |
| `transport` | `'websocket'` | `'websocket'` (crossws and partysocket), `'socket.io'` (rooms, adapters) or `'sse'` (one-way stream over HTTP, control messages over POST). |
| `path` | `'/ws'`, or `'/events'` for `sse` | Where the server listens. The client connects to the same origin at this path. |
| `url` | same origin at `path` | A full URL for the client, such as `wss://realtime.acme.com/ws`, when the realtime server is on another host. |
| `heartbeat` | `25000` | Declared for a client ping interval in milliseconds. The built-in clients do not read it yet. |

Realtime adds no environment variables.

## Push your first update

Three files make a live project name: a channel both ends import, an emit on the server, and a subscription in a component.

Define the channel. `zod` is used for schemas here (`pnpm add zod`); any [Standard Schema](https://standardschema.dev) library works.

```ts
// app/channels/projects.ts
import { defineChannel } from 'kitr/realtime'
import { z } from 'zod'

export const projectSchema = z.object({ id: z.string(), name: z.string(), teamId: z.string() })
export type Project = z.infer<typeof projectSchema>

export const projectChannel = defineChannel({
  name: 'projects',
  params: z.object({ id: z.string() }),
  events: {
    updated: {
      schema: projectSchema,
      effects: (project, { id }) => ({ set: [{ key: ['projects', 'detail', id], data: project }] }),
    },
  },
})
```

Add the channel to the server in `server/realtime.ts`, then emit from the controller that renames a project:

```ts
// server/realtime.ts
import { createRealtimeServer } from 'kitr/realtime/server'
import { projectChannel } from '#app/channels/projects'

export const realtime = createRealtimeServer({
  channels: [projectChannel],
})
```

```ts
// server/controllers/projects_controller.ts
import type { HttpContext } from 'kitr/api'
import { type Project, projectChannel } from '#app/channels/projects'
import { realtime } from '#server/realtime'

export default class ProjectsController {
  async update({ params, body }: HttpContext<{ id: string }, Record<string, string>, { name: string }>) {
    const project: Project = { id: params.id, name: body.name, teamId: 'team_acme' }
    // ... save it with your database
    realtime.use(projectChannel, { id: params.id }).emit('updated', project)
    return project
  }
}
```

Subscribe in the component that shows the project:

```tsx
// app/features/projects/project-title.tsx
import { useQuery } from 'kitr/query'
import { useChannel } from 'kitr/realtime/react'
import { type Project, projectChannel } from '#app/channels/projects'
import { realtime } from '#app/realtime'

export function ProjectTitle({ id }: { id: string }) {
  useChannel(realtime(), projectChannel, { id })
  const { data } = useQuery<Project>({ queryKey: ['projects', 'detail', id], enabled: false })
  return <h1>{data?.name ?? 'Loading'}</h1>
}
```

When Grace saves "Apollo launch", the server sends one `updated` event to every connection subscribed to `projects` with `{ id: 'prj_1' }`. Each client validates the payload against `projectSchema` and runs `effects`, which writes the project into the `['projects', 'detail', 'prj_1']` cache entry. Ada's and Linus's headings re-render with the new name. No refetch happens.

## Define a channel

A channel is a name, the params a subscriber picks, and the events the server sends. One file under `app/channels/` is imported by both ends, so the event names and payload types cannot drift.

```ts
// app/channels/projects.ts
import { defineChannel } from 'kitr/realtime'
import { z } from 'zod'

export 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().trim() }),
  events: {
    updated: { schema: projectSchema },
    deleted: {},
    typing: { schema: z.object({ user: z.string() }) },
  },
})
```

| Field | Runs on | Description |
| --- | --- | --- |
| `name` | both | Unique channel name. Two channels with the same name collide. |
| `params` | server | Schema for what a client subscribes with. Invalid params fail with `VALIDATION`. |
| `events` | both | Event names, each with an optional `schema` and `effects`. |
| `authorize(params, ctx)` | server | Return `false` to refuse a subscription. See Authorize a subscription. |
| `onMessage(event, data, params, ctx)` | server | Receives events a client sends up the channel. |

`kitr make:channel <name>` writes a starting file (see Commands). Without a `params` schema, give the params type as a type argument: `defineChannel<{ room: string }>({ ... })`.

## Validate params and payloads

On the server, `params` runs before anything else. Its output, not the raw input, is what the subscription is filed under. A schema that trims or lowercases therefore 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; the subscription's error handler gets a `VALIDATION` error instead:

```ts
// app/features/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**
> The server does not validate what you pass to `emit`. Payload schemas protect clients from a malformed event; they do not stop the server from sending one. 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 and joins stay on the server. Don't call `queryClient.setQueryData` in a component for socket events; declare it in `effects` so every subscriber behaves the same.

## Emit from the server

On the server, `realtime.use(channel, params)` returns an emitter for one topic. Import `realtime` from `server/realtime.ts` in a controller, an action or a job:

```ts
// app/actions/projects/rename.ts
import { defineAction } from 'kitr/actions'
import { z } from 'zod'
import { projectChannel } from '#app/channels/projects'
import { realtime } from '#server/realtime'

export const renameProject = defineAction('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
    realtime.use(projectChannel, { id: input.id }).emit('updated', project)
    return project
  },
})
```

The emitter also reports how many connections are listening, which helps you skip expensive work nobody will see:

```ts
// server/jobs/project_stats.ts
import { projectChannel } from '#app/channels/projects'
import { realtime } from '#server/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()` above, `use(projectChannel, { id: 'prj_1' })` reaches a client that subscribed with `' prj_1 '`; `use(projectChannel, { id: ' prj_1 ' })` reaches nobody. Key order in params does not matter: `{ a: 1, b: 2 }` and `{ b: 2, a: 1 }` are the same topic.

## Subscribe in a component

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/features/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 <p>{live.status === 'open' ? 'Live' : live.status}</p>
}
```

`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. `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 kitr subscription contract, so `useSubscription` and `defineSubscription` from [query](/packages/query) take it directly. Use this when your queries file already holds the other contracts for a feature:

```ts
// app/features/projects/queries.ts
import { defineSubscription } 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 = () =>
  defineSubscription(channel(realtime(), projectChannel).subscription())
```

```tsx
// app/features/projects/project-live.tsx
import { useSubscription } from 'kitr/query'
import { useMemo } from 'react'
import { projectSubscription } from './queries'

export function ProjectLive({ id }: { id: string }) {
  const subscription = useMemo(() => projectSubscription(), [])
  const live = useSubscription(subscription, { id })
  return <span>{live.status}</span>
}
```

The handle's `key(params)` is `['realtime', 'projects', { id: 'prj_1' }]`. It names the subscription; events still write to the keys your `effects` return.

## 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 <span>{status === 'open' ? 'Online' : 'Reconnecting'}</span>
}
```

The WebSocket client reconnects on its own with backoff and sends every subscription again after it reconnects. Events emitted while a client was offline are not replayed, so pair live events with queries: on reconnect, invalidate what matters, or let `refetchOnWindowFocus` catch up.

## Send events from the client

A client can send an event up a channel with `channel(...).send(params, event, data)`. The server runs the channel's `params` schema and `authorize`, then calls the channel's `onMessage`. Here, typing in a project's comment box tells everyone else on the board.

The channel declares the `typing` 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. The server matches channels by `name`, so the client keeps importing the shared file and the handler never reaches the browser:

```ts
// server/realtime.ts
import { createRealtimeServer } from 'kitr/realtime/server'
import { projectChannel } from '#app/channels/projects'

export const realtime = createRealtimeServer({
  channels: [
    {
      ...projectChannel,
      onMessage: (event, data, params) => {
        if (event !== 'typing' || typeof (data as { user?: unknown })?.user !== 'string') return
        realtime.use(projectChannel, params as { id: string }).emit('typing', data as { user: string })
      },
    },
  ],
})
```

```tsx
// app/features/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 <textarea onKeyDown={() => project.send({ id: projectId }, 'typing', { user: 'Ada' })} />
}
```

`onMessage(event, data, params, ctx)` receives the validated params and the connection's context. Check `data` yourself, as above: event schemas are checked when events arrive at a client, not at the server. For anything that changes data, call an action or an API route instead; `send` has no reply and no retry.

## Authenticate connections

On the server, `authorize(request)` runs once, when a connection opens. Return the connection's context, such as the signed-in user. Return `false` and the upgrade is refused with `401 Unauthorized`. Throw a `KitrError` to refuse with its status; any other thrown error becomes a 401.

```ts
// server/realtime.ts
import { KitrError } from 'kitr/app'
import { createRealtimeServer } from 'kitr/realtime/server'
import { projectChannel } from '#app/channels/projects'
import { userFromSession } from '#server/auth'

export const realtime = createRealtimeServer({
  channels: [projectChannel],
  authorize: async (request) => {
    const cookie = request.headers.get('cookie') ?? ''
    const session = cookie.match(/(?:^|;\s*)acme_session=([^;]+)/)?.[1]
    const user = session ? await userFromSession(session) : null
    if (!user) throw KitrError.unauthorized()
    return { user }
  },
})
```

`userFromSession` stands for your own lookup. Browsers send cookies with the WebSocket upgrade for the same site, so a session cookie works without extra client code. For a token, put it in the URL from the client and read it with `new URL(request.url).searchParams.get('token')`:

```ts
// app/realtime.ts
import { createSocketClient, type SocketTransport } from 'kitr/realtime'
import realtimeConfig from '#config/realtime'

let transport: SocketTransport | undefined

export function realtime(): SocketTransport {
  transport ??= createSocketClient({
    url: async () => {
      const token = localStorage.getItem('acme_token') ?? ''
      return `${window.location.origin.replace(/^http/, 'ws')}${realtimeConfig.path}?token=${token}`
    },
  })
  return transport
}
```

The `url` function runs again on every reconnect, so a refreshed token is picked up.

## Authorize a subscription

`authorize(params, ctx)` on the channel decides who may subscribe to which topic. `ctx` is what the server's `authorize(request)` returned. It also guards `send`.

```ts
// app/channels/projects.ts
import { defineChannel } from 'kitr/realtime'
import { z } from 'zod'

type Ctx = { user: { id: string; teamIds: string[] } }

export const projectChannel = defineChannel({
  name: 'projects',
  params: z.object({ id: z.string(), teamId: z.string() }),
  events: { updated: {} },
  authorize: ({ teamId }, ctx) => (ctx as Ctx).user.teamIds.includes(teamId),
})
```

When Linus, who is not on `team_acme`, subscribes with `{ id: 'prj_1', teamId: 'team_acme' }`, his `useChannel` gets `error: { code: 'FORBIDDEN', message: 'Not allowed on channel "projects"', status: 403 }` and receives no events. A channel without `authorize` lets every authenticated connection subscribe.

## Handle errors

Errors reach the most specific handler. Subscription errors (bad params, `FORBIDDEN`, an unknown channel name) go to that subscription's `onError`. Errors with no subscription, such as a `send` on an unknown channel, go to the transport's `onError`:

```ts
// app/realtime.ts
import { createSocketClient, type SocketTransport } from 'kitr/realtime'
import realtimeConfig from '#config/realtime'

let transport: SocketTransport | undefined

export function realtime(): SocketTransport {
  transport ??= createSocketClient({
    url: () => `${window.location.origin.replace(/^http/, 'ws')}${realtimeConfig.path}`,
    reconnect: { maxRetries: 20, minReconnectionDelay: 1_000, maxReconnectionDelay: 30_000 },
    onError: (error) => console.warn('[realtime]', error.code, error.message),
  })
  return transport
}
```

This is the same file `kitr add` writes, with `createSocketClient` called directly so you can pass the options it takes:

| Option | Default | Description |
| --- | --- | --- |
| `url` | required | `ws://` or `wss://` URL, or a function returning one (sync or async). |
| `protocols` | none | WebSocket subprotocols. |
| `reconnect.maxRetries` | partysocket's default | Give up after this many attempts. |
| `reconnect.minReconnectionDelay` | partysocket's default | First retry delay in milliseconds. |
| `reconnect.maxReconnectionDelay` | partysocket's default | Longest retry delay in milliseconds. |
| `onError(error)` | none | Server errors not tied to a subscription. |

See [partysocket](https://docs.partykit.io/reference/partysocket-api/) for the reconnect defaults.

## Connect from another origin

When the realtime server runs on its own host, set `url` in config. The client uses it as is, instead of the page's origin plus `path`:

```ts
// config/realtime.ts
import { defineConfig } from 'kitr/realtime'

export default defineConfig({
  transport: 'websocket',
  path: '/ws',
  url: 'wss://realtime.acme.com/ws',
})
```

From a Node script or a test, there is no page origin. Call `createSocketClient` with a full URL; Node 22 and later ship a global `WebSocket`:

```ts
// scripts/watch_project.ts
import { channel, createSocketClient } from 'kitr/realtime'
import { projectChannel } from '#app/channels/projects'

const transport = createSocketClient({ url: 'ws://localhost:3001/ws' })
channel(transport, projectChannel)
  .subscription()
  .subscribe({ id: 'prj_1' }, { next: (event) => console.log(event) })
// => { event: 'updated', data: { id: 'prj_1', name: 'Apollo launch', teamId: 'team_acme' } }
```

## Bind the server in the container

`server/realtime.ts` exports one module-level server, which is all most apps need. To reach it through the app container instead, as `ctx.app.container.use('realtime')`, register `realtimeProvider` and attach the bound server after boot:

```ts
// server/main.ts
import { mountRoutes } from 'kitr/api'
import { realtimeProvider } from 'kitr/realtime/server'
import { createServer } from 'kitr/server'
import { projectChannel } from '#app/channels/projects'
import { config } from '../config'
import rc from '../kitrrc'
import { routes } from './routes'

export const server = createServer({
  providers: [...rc.providers, realtimeProvider({ channels: [projectChannel] })],
  config,
})
mountRoutes(server, routes)
await server.boot()
server.app.container.use('realtime').attach(server, '/ws')
```

```ts
// server/controllers/projects_controller.ts
import type { HttpContext } from 'kitr/api'
import { projectChannel } from '#app/channels/projects'

export default class ProjectsController {
  async destroy({ app, params }: HttpContext<{ id: string }>) {
    app.container.use('realtime').use(projectChannel, { id: params.id }).emit('deleted', null)
    return { ok: true }
  }
}
```

The provider only binds the server; it does not register the route, and its `path` option is not used. `attach` must run before `server.listen()`, because it turns on WebSocket support for the listener. For SSE, `sseProvider` from `kitr/realtime/sse/server` does the same.

## Transports

All three transports implement one client interface and share the wire protocol, so channels, `effects`, `useChannel` and `emit` are identical. What changes is the server object in `server/realtime.ts`, its `attach` call, and the client the app creates.

| `transport` | Client | Server | Use it when |
| --- | --- | --- | --- |
| `websocket` | `createSocketClient` (partysocket) | `createRealtimeServer` (crossws via h3) | The default. Two-way, low latency, runs on Node, Bun, Deno and Cloudflare. |
| `socket.io` | `createSocketIoClient` | `attachSocketIo` | You need several server instances (Redis adapter) or already run Socket.IO. |
| `sse` | `createSseTransport` (EventSource plus POST) | `createSseServer` | Your host or proxy does not allow WebSocket upgrades. |

### WebSocket

The default. `kitr add realtime` sets it up: `crossws` on the server, `partysocket` in the browser.

```ts
// server/realtime.ts
import { createRealtimeServer } from 'kitr/realtime/server'
import { projectChannel } from '#app/channels/projects'

export const realtime = createRealtimeServer({ channels: [projectChannel] })
```

```ts
// server/main.ts
import { mountRoutes } from 'kitr/api'
import { createServer } from 'kitr/server'
import { config } from '../config'
import rc from '../kitrrc'
import { realtime } from './realtime'
import { routes } from './routes'

export const server = createServer({ providers: rc.providers, config })
mountRoutes(server, routes)
// WebSocket upgrades on /api/ws, next to the routes.
realtime.attach(server, '/api/ws')
```

`attach(server, path)` registers the upgrade route on the server's h3 app and turns on `websocket` for `server.listen()`. Without `crossws` installed, `listen()` throws `[kitr/server] websocket: true needs the crossws package. Run kitr add socket.` See the [crossws docs](https://crossws.unjs.io) for runtime adapters.

### Socket.IO

Install both halves:

```sh
pnpm add socket.io socket.io-client
```

Set the transport and path in config:

```ts
// config/realtime.ts
import { defineConfig } from 'kitr/realtime'

export default defineConfig({
  transport: 'socket.io',
  path: '/api/socket.io',
})
```

On the server, create a Socket.IO server and serve your channels on it with `attachSocketIo`. Its `authorize` receives the socket and reads `socket.handshake.auth`:

```ts
// server/realtime.ts
import { KitrError } from 'kitr/app'
import { attachSocketIo } from 'kitr/realtime/socket.io/server'
import { Server } from 'socket.io'
import { projectChannel } from '#app/channels/projects'
import { userFromToken } from '#server/auth'

export const io = new Server({ path: '/api/socket.io' })

export const realtime = attachSocketIo(io, {
  channels: [projectChannel],
  authorize: async (socket) => {
    const user = await userFromToken(String(socket.handshake.auth.token ?? ''))
    if (!user) throw KitrError.unauthorized()
    return { user }
  },
})
```

Bind it to the HTTP server once the kitr server is listening:

```ts
// server/main.ts
import type { Server as HttpServer } from 'node:http'
import { createServer } from 'kitr/server'
import { config } from '../config'
import rc from '../kitrrc'
import { io } from './realtime'

export const server = createServer({ providers: rc.providers, config })
const listening = await server.listen({ port: 3001 })
io.attach(listening.node as HttpServer)
```

In the browser, `createRealtimeClient` refuses `socket.io` so that `socket.io-client` stays out of bundles that don't use it. Create the client from `kitr/realtime/socket.io`, which also exports `channel` and `defineChannel`:

```ts
// app/realtime.ts
import type { SocketTransport } from 'kitr/realtime'
import { createSocketIoClient } from 'kitr/realtime/socket.io'
import realtimeConfig from '#config/realtime'

let transport: SocketTransport | undefined

export function realtime(): SocketTransport {
  transport ??= createSocketIoClient({
    path: realtimeConfig.path,
    auth: { token: localStorage.getItem('acme_token') ?? '' },
  })
  return transport
}
```

`createSocketIoClient` takes every [socket.io-client option](https://socket.io/docs/v4/client-options/) plus `url` and `onError`, and exposes the raw socket as `.socket`. Each channel topic is a Socket.IO room, so `realtime.use(channel, params).emit()` emits to that room. Pass `namespace: '/board'` to `attachSocketIo` to serve channels on a namespace. The Socket.IO emitter has no `size`.

### Server-Sent Events

SSE needs no extra packages. The server streams events over one long `GET`; the client subscribes and sends with `POST`s to the same path.

```ts
// config/realtime.ts
import { defineConfig } from 'kitr/realtime'

export default defineConfig({
  transport: 'sse',
  path: '/api/events',
})
```

```ts
// server/realtime.ts
import { createSseServer } from 'kitr/realtime/sse/server'
import { projectChannel } from '#app/channels/projects'

export const realtime = createSseServer({
  channels: [projectChannel],
  keepAlive: 15_000,
})
```

```ts
// server/main.ts
import { mountRoutes } from 'kitr/api'
import { createServer } from 'kitr/server'
import { config } from '../config'
import rc from '../kitrrc'
import { realtime } from './realtime'
import { routes } from './routes'

export const server = createServer({ providers: rc.providers, config })
mountRoutes(server, routes)
realtime.attach(server, '/api/events')
```

The `app/realtime.ts` that `kitr add` wrote already works: `createRealtimeClient` returns `createSseTransport({ url: config.url ?? config.path })` for `sse`. To pass headers on the control `POST`s, create the transport yourself:

```ts
// app/realtime.ts
import { createSseTransport, type SocketTransport } from 'kitr/realtime'

let transport: SocketTransport | undefined

export function realtime(): SocketTransport {
  transport ??= createSseTransport({
    url: '/api/events',
    headers: () => ({ 'x-acme-team': 'team_acme' }),
  })
  return transport
}
```

| Option | Default | Description |
| --- | --- | --- |
| `keepAlive` (server) | `20000` | Milliseconds between keep-alive comments, so proxies don't close an idle stream. |
| `authorize(request)` (server) | none | Runs when the stream opens. Same rules as the WebSocket server. |
| `url` (client) | required | The stream URL. Control messages are POSTed to it. |
| `headers` (client) | none | Headers for the control `POST`s, or a function returning them. |
| `onError` (client) | none | Errors without a subscription, and failed `POST`s as `NETWORK`. |

`EventSource` cannot set headers, so the stream authenticates by cookie (it is opened with credentials) or by a `?token=` in the URL. With SSE, subscribe with params already in the form the `params` schema outputs: the SSE server does not send normalised params back, so a subscription with `' prj_1 '` receives nothing. See [MDN on Server-Sent Events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events).

## Run more than one server instance

Each server keeps its subscriptions in memory. With two instances behind a load balancer, an `emit` on instance A reaches only the clients connected to A.

> **Warning**
> With the WebSocket or SSE transport and several instances, clients on other instances silently miss events. Run one realtime instance, or use Socket.IO with an adapter.

With Socket.IO, add the [Redis adapter](https://socket.io/docs/v4/redis-adapter/) and every instance's emits reach every room:

```ts
// server/realtime.ts
import { createAdapter } from '@socket.io/redis-adapter'
import { attachSocketIo } from 'kitr/realtime/socket.io/server'
import { createClient } from 'redis'
import { Server } from 'socket.io'
import { projectChannel } from '#app/channels/projects'

const pub = createClient({ url: process.env.REDIS_URL })
const sub = pub.duplicate()
await Promise.all([pub.connect(), sub.connect()])

export const io = new Server({ path: '/api/socket.io', adapter: createAdapter(pub, sub) })
export const realtime = attachSocketIo(io, { channels: [projectChannel] })
```

Socket.IO also needs sticky sessions at the load balancer unless clients use only the `websocket` transport (`transports: ['websocket']` on the client).

## Commands

| Command | What it does |
| --- | --- |
| `kitr make:channel <name>` | Writes `app/channels/<name>.ts` with `defineChannel`, a `zod` params schema, `updated` and `deleted` events and an open `authorize`. |

```sh
kitr make:channel invoices
```

```ts
// app/channels/invoices.ts
import { defineChannel } from 'kitr/realtime'
import { z } from 'zod'

export const invoicesChannel = defineChannel({
  name: 'invoices',
  params: z.object({ id: z.string() }),
  events: {
    updated: {
      effects: (data: unknown, { id }) => ({ set: [{ key: ['invoices', 'detail', id], data }] }),
    },
    deleted: {
      effects: () => ({ invalidate: [['invoices', 'list']] }),
    },
  },
  authorize: () => true,
})
```

The file name is the snake-case form of the name. Add the new channel to `channels` in `server/realtime.ts`; a client that subscribes to a channel the server does not list gets `NOT_FOUND`.

## Errors

Errors travel as kitr error shapes (`{ code, message, status }`) to the subscription's `onError`, or to the transport's `onError` when there is no subscription.

| Code | When | What to do |
| --- | --- | --- |
| `NOT_FOUND` | `Channel "invoices" not found`: the channel is not in the server's `channels`. | Add it to `createRealtimeServer({ channels })`. |
| `VALIDATION` (server) | The subscription params fail the channel's `params` schema. `fieldErrors` says which field. | Fix the params the component passes. |
| `VALIDATION` (client) | `Invalid payload for projects.updated`: an emitted payload fails the event `schema`. | Fix the data the server emits. |
| `FORBIDDEN` | The channel's `authorize` returned `false`: `Not allowed on channel "projects"`. | Expected for users without access; show a message. |
| HTTP `401` | The server's `authorize(request)` returned `false` or threw. The connection never opens. | Sign in, or send the cookie or token. |
| HTTP `410` | SSE only: a control `POST` named a connection the server no longer has. | The client reconnects and resubscribes on its own. |
| `NETWORK` | SSE only: a control `POST` failed. | Check the network; the subscription is not active. |
| `[kitr/realtime] transport "socket.io": create the client with createSocketIoClient() from kitr/realtime/socket.io` | `createRealtimeClient` with `transport: 'socket.io'`. | Use `createSocketIoClient` in `app/realtime.ts`. |

## Testing

Server and client both run in Node, so a test can start the real server on a free port and subscribe with the real client. This checks the channel, the authorization and the emit together:

```ts
// test/project_channel.test.ts
import { channel, createSocketClient, type SocketTransport } from 'kitr/realtime'
import { createRealtimeServer } from 'kitr/realtime/server'
import { silentLogger } from 'kitr/logger'
import { createServer } from 'kitr/server'
import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
import { projectChannel } from '#app/channels/projects'

describe('projects channel', () => {
  const server = createServer({ logger: silentLogger })
  const realtime = createRealtimeServer({
    channels: [projectChannel],
    authorize: (request) => ({ user: new URL(request.url).searchParams.get('user') }),
  })
  realtime.attach(server, '/ws')
  let base = ''
  let close: () => Promise<void>
  let client: SocketTransport

  beforeAll(async () => {
    const listening = await server.listen({ port: 0, host: '127.0.0.1' })
    base = listening.url.replace(/^http/, 'ws')
    close = listening.close
  })
  afterAll(async () => {
    client?.close()
    await close()
  })

  it('delivers updated projects to subscribers', async () => {
    client = createSocketClient({ url: `${base}/ws?user=ada`, reconnect: { maxRetries: 0 } })
    const events: unknown[] = []
    channel(client, projectChannel)
      .subscription()
      .subscribe({ id: 'prj_1' }, { next: (event) => events.push(event) })

    await vi.waitFor(() => expect(realtime.use(projectChannel, { id: 'prj_1' }).size).toBe(1))
    realtime
      .use(projectChannel, { id: 'prj_1' })
      .emit('updated', { id: 'prj_1', name: 'Apollo launch', teamId: 'team_acme' })

    await vi.waitFor(() =>
      expect(events).toEqual([
        { event: 'updated', data: { id: 'prj_1', name: 'Apollo launch', teamId: 'team_acme' } },
      ]),
    )
  })
})
```

`port: 0` picks a free port, and `vi.waitFor` retries until the subscription has reached the server. The test uses the channel from "Push your first update".

To test what the cache looks like after an event, call the subscription's `effects` directly. It is a plain function:

```ts
// test/project_effects.test.ts
import { channel, type SocketTransport } from 'kitr/realtime'
import { expect, it } from 'vitest'
import { projectChannel } from '#app/channels/projects'

it('writes the project detail', () => {
  const idle = { subscribe: () => () => {} } as unknown as SocketTransport
  const effects = channel(idle, projectChannel).subscription().effects!
  const project = { id: 'prj_1', name: 'Apollo launch', teamId: 'team_acme' }
  expect(effects({ event: 'updated', data: project }, { id: 'prj_1' })).toEqual({
    set: [{ key: ['projects', 'detail', 'prj_1'], data: project }],
  })
})
```

For component tests, pass a memory transport (see Extend) instead of `realtime()`, render with `renderWithKitr` from [testing](/packages/testing), and push events from the test.

## Extend

Every client is a `SocketTransport`. Implement it to put channels on another service, such as a hosted pub/sub, or to drive components from a test. This one keeps subscriptions in memory and lets the caller push events:

```ts
// test/memory-transport.ts
import type { SocketStatus, SocketTransport } from 'kitr/realtime'
import { topicFor } from 'kitr/realtime'

type Handler = (event: string, data: unknown) => void

export function createMemoryTransport() {
  const topics = new Map<string, Set<Handler>>()
  const sent: Array<{ channel: string; params: unknown; event: string; data: unknown }> = []
  const transport: SocketTransport & {
    push(channel: string, params: unknown, event: string, data: unknown): void
    sent: typeof sent
  } = {
    status: 'open' as SocketStatus,
    onStatus: () => () => {},
    subscribe(channel, params, onEvent) {
      const topic = topicFor(channel, params)
      if (!topics.has(topic)) topics.set(topic, new Set())
      topics.get(topic)!.add(onEvent)
      return () => topics.get(topic)?.delete(onEvent)
    },
    send(channel, params, event, data) {
      sent.push({ channel, params, event, data })
    },
    close() {
      topics.clear()
    },
    push(channel, params, event, data) {
      for (const handler of topics.get(topicFor(channel, params)) ?? []) handler(event, data)
    },
    sent,
  }
  return transport
}
```

Use it where the app expects a transport. `channel()`, `useChannel` and `useRealtimeStatus` only call these five members:

```tsx
// app/features/projects/project-title.test.tsx
import { act, screen } from '@testing-library/react'
import { queryProvider } from 'kitr/query'
import { useChannel } from 'kitr/realtime/react'
import { createTestApp } from 'kitr/testing'
import { renderWithKitr } from 'kitr/testing/react'
import { expect, it } from 'vitest'
import { projectChannel } from '#app/channels/projects'
import { createMemoryTransport } from '../../../test/memory-transport'

const transport = createMemoryTransport()

function LastProjectName({ id }: { id: string }) {
  const live = useChannel(transport, projectChannel, { id })
  return <p>{live.lastEvent?.event === 'updated' ? live.lastEvent.data.name : 'Waiting'}</p>
}

it('shows the renamed project', async () => {
  const app = await createTestApp({ providers: [queryProvider] })
  renderWithKitr(<LastProjectName id="prj_1" />, { app })
  act(() =>
    transport.push('projects', { id: 'prj_1' }, 'updated', {
      id: 'prj_1',
      name: 'Apollo launch',
      teamId: 'team_acme',
    }),
  )
  expect(await screen.findByText('Apollo launch')).toBeTruthy()
})
```

| Member | Description |
| --- | --- |
| `subscribe(channel, params, onEvent, onError?)` | Start receiving events for a topic. Returns the unsubscribe function. |
| `send(channel, params, event, data)` | Send a client event to the server. |
| `close()` | Close the connection. |
| `status` | `'connecting'`, `'open'` or `'closed'`. |
| `onStatus(listener)` | Listen for status changes. Returns the remove function. |

A real transport also resubscribes after reconnecting and calls `onError` with a kitr error shape. Read `createSseTransport` in the source for a short complete example, and see [`SocketTransport`](/packages/realtime/api) in the API reference.

## Next steps

- [API reference](/packages/realtime/api): every export of the seven entry points.
- [query](/packages/query): the cache that `effects` write to, and `useSubscription`.
- [server](/packages/server): `createServer`, `listen` and shutdown.
- [crossws](https://crossws.unjs.io) and [Socket.IO](https://socket.io/docs/v4/): the libraries underneath.
- [Getting started](/docs/getting-started): the `realtime` kit, a chat room built on these pieces.