realtime

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 WebSockets by default, or Socket.IO or Server-Sent Events, and the channel code is the same on all three. The server half (@kitr/realtime/server) emits; the client half (@kitr/realtime, @kitr/realtime/react) subscribes. For one-off server calls that return a value, use actions or api. The examples follow Acme's project board at app.acme.com.

Install

To preview the changes:

Configure

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

// 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'.

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. The examples use zod (pnpm add zod); any Standard Schema library works.

// 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 }] }),
    },
  },
})

Serve the channel from the server route table, then emit from the controller that renames a project:

// server/routes.ts
import { defineRoutes } from '@kitr/kitr/app'
import api from '@kitr/api/routes'
import realtime from '@kitr/realtime/routes'
import { projectChannel } from '#app/channels/projects'
import ProjectsController from '#server/controllers/projects-controller'

export const routes = defineRoutes([api, realtime], (root) =>
  root.group('/api', (r) =>
    r.resource('projects', ProjectsController).ws('/ws', { channels: [projectChannel] }),
  ),
)

export type ServerRoutes = typeof routes.$types

.ws(path, options) is a route like any other: it mounts the WebSocket server at /api/ws and binds it in the container as realtime, so controllers, actions and jobs reach it through app.container.use('realtime'):

// server/controllers/projects-controller.ts
import type { HttpContext } from '@kitr/api'
import { type Project, projectChannel } from '#app/channels/projects'

export default class ProjectsController {
  async update({ app, 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
    app.container.use('realtime').use(projectChannel, { id: params.id }).emit('updated', project)
    return project
  }
}

Subscribe in the component that shows the project:

// app/components/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 a project is saved, the server sends one updated event to every connection subscribed with that id. Each client validates the payload and runs effects, which writes the project into the ['projects', 'detail', id] cache entry. Headings re-render with the new name, with no refetch.

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 names and payload types cannot drift.

// 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() }) },
  },
})

kitr make:channel <name> 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:

// 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.

Update the cache from events

effects(data, params) returns the same cache effect a query mutation returns: set writes an entry, invalidate marks entries stale so mounted queries refetch, remove drops them.

// 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:

// 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:

// 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 provider, which every kit with React includes.

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

useChannel returns { status, lastEvent, error }:

Changing params resubscribes.

Subscribe with useSubscription

channel(transport, definition) binds a channel to a transport. Its subscription() is a subscription contract that useSubscription and subscriptionFrom from query take directly:

// 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())

// 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 <span>{live.status}</span>
}

Show the connection status

useChannel reports the subscription. To show whether the socket itself is connected, use useRealtimeStatus(transport):

// 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 with backoff and resubscribes.

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:

// 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:

// 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 })
      },
    },
  ],
})

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

onMessage(event, data, params, ctx, realtime) receives the validated params, the connection's context and the server, to emit back. Check data yourself: event schemas are checked at the client, not the server.

Authenticate connections

authorize(request) runs once, when a connection opens. Return the connection's context, such as the signed-in user. Return false to refuse with 401, or throw a KitrError to refuse with its status.

// server/routes.ts
import { KitrError } from '@kitr/kitr/app'
import { projectChannel } from '#app/channels/projects'
import { userFromSession } from '#server/auth'

r.ws('/ws', {
  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 same-site WebSocket upgrade, so a session cookie needs no client code. For a token, put it in the URL and read it with new URL(request.url).searchParams.get('token'):

// 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.

Close a revoked session

Nothing re-checks an open connection, so a user who signs out everywhere or is banned keeps receiving the topics they already joined until their socket drops. closeSessions(userId) closes every connection of the user on this instance and returns how many:

// server/revoke.ts
import { realtime } from '#server/realtime'

export function revokeLiveAccess(userId: string) {
  return realtime.closeSessions(userId)
}

A WebSocket closes with code 4401; the client reconnects and authorize(request) refuses the revoked session. Connections belong to a user by the context authorize returned (userId, user.id or id); if yours is shaped differently, say how with identify:

// server/routes.ts
r.ws('/ws', {
  channels: [projectChannel],
  authorize: async (request) => ({ account: await accountFromSession(request) }),
  identify: (ctx) => (ctx as { account: { id: string } }).account.id,
})

The session controller from @kitr/auth calls a hook after it revokes every session of a user, so one line connects the two:

// server/controllers/auth_controller.ts
import { createSessionController } from '@kitr/auth/server'
import { realtime } from '#server/realtime'
import { sessions, users } from '#server/auth'

export default createSessionController({
  sessions,
  users,
  onSessionsRevoked: (userId) => realtime.closeSessions(userId),
})

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.

// 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),
})

A user outside the team gets error: { code: 'FORBIDDEN', ... status: 403 } from useChannel and receives no events. A channel without authorize lets every authenticated connection subscribe.

Handle errors

Subscription errors (bad params, FORBIDDEN, an unknown channel) go to that subscription's onError; errors with no subscription go to the transport's onError:

// 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 file kitr add writes, with createSocketClient called directly so you can pass its options:

See partysocket for the reconnect defaults.

Connect from another origin

When the realtime server runs on its own host, set url in config:

// 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, call createSocketClient with a full URL (Node 22 and later ship a global WebSocket):

// 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' } }

The server in the container

The first .ws() or .sse() link is available as realtime in the container. To run a server outside the route table, use createRealtimeServer(options) from @kitr/realtime/server and call attach(server, path) before server.listen().

Transports

All three transports share one client interface, so channels, effects, useChannel and emit are identical. Only the server link (.ws, .sse or an attached Socket.IO server) and the client the app creates change.

WebSocket

The default, set up by kitr add realtime: crossws on the server, partysocket in the browser.

// server/routes.ts
import { defineRoutes } from '@kitr/kitr/app'
import realtime from '@kitr/realtime/routes'
import { projectChannel } from '#app/channels/projects'
import realtimeConfig from '../config/realtime'

export const routes = defineRoutes([realtime], (r) =>
  // WebSocket upgrades on /api/ws, next to the routes.
  r.ws('/api/ws', { channels: [projectChannel] }),
)

// server/main.ts
import { createServer } from '@kitr/server'
import { routes } from './routes'
import { config } from '../config'
import rc from '../kitrrc'

export const server = createServer({ providers: rc.providers, config, routes })

.ws(path, options) takes channels, authorize, adapter and the other createRealtimeServer options. Without crossws installed, listen() throws and tells you to run kitr add socket. See the crossws docs for runtime adapters.

Socket.IO

pnpm add socket.io socket.io-client

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

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

// server/realtime.ts
import { KitrError } from '@kitr/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 }
  },
})

// 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'
import { routes } from './routes'

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

// 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
}

Server-Sent Events

SSE needs no extra packages. The server streams events over one long GET; the client sends with POSTs to the same path. Set the transport in config:

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

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

// server/routes.ts
r.sse('/api/events', { channels: [projectChannel], keepAlive: 15_000 })

.sse(path, options) mounts the stream and its control endpoint on one path. The app/realtime.ts that kitr add wrote already works; to pass headers on the control POSTs, create the transport yourself:

// 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
}

EventSource cannot set headers, so the stream authenticates by cookie or by a ?token= in the URL. See MDN on Server-Sent Events.

Run more than one server instance

Each instance keeps its own connections. To make an emit on one instance reach clients on another, give the realtime server an adapter, the shared bus instances publish through. The default adapter is in memory; the Redis adapter reaches every instance connected to the same Redis.

Install an ioredis client:

pnpm add ioredis

Create the adapter once, in its own file, and pass it to every realtime link on the route table:

// server/realtime_adapter.ts
import { Redis } from 'ioredis'
import { createRedisRealtimeAdapter } from '@kitr/realtime/server'

export const realtimeAdapter = createRedisRealtimeAdapter(
  new Redis(process.env.REDIS_URL ?? 'redis://localhost:6379'),
  { prefix: 'acme-app:rt:' },
)

// server/routes.ts
import { defineRoutes } from '@kitr/kitr/app'
import realtime from '@kitr/realtime/routes'
import { projectChannel } from '#app/channels/projects'
import { realtimeAdapter } from './realtime_adapter'

export const routes = defineRoutes([realtime], (r) =>
  r.ws('/api/ws', { channels: [projectChannel], adapter: realtimeAdapter }),
)

Add the connection string to .env:

REDIS_URL=redis://localhost:6379

Nothing else changes: an emit on any instance is published through Redis and every instance with a subscriber writes it to its own sockets.

With Socket.IO, use Socket.IO's own Redis adapter instead:

// 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 set transports: ['websocket'].

Commands

// 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,
})

Add the new channel to the channels of the .ws() link in server/routes.ts, or subscribers get NOT_FOUND.

Errors

Errors travel as { code, message, status } shapes to onError.

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:

// test/project-channel.test.ts
import { defineRoutes } from '@kitr/kitr/app'
import { silentLogger } from '@kitr/logger'
import { channel, createSocketClient, type SocketTransport } from '@kitr/realtime'
import ws from '@kitr/realtime/routes'
import { createServer } from '@kitr/server'
import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
import { projectChannel } from '#app/channels/projects'

describe('projects channel', () => {
  const routes = defineRoutes([ws], (r) =>
    r.ws('/ws', {
      channels: [projectChannel],
      authorize: (request) => ({ user: new URL(request.url).searchParams.get('user') }),
    }),
  )
  const server = createServer({ logger: silentLogger, routes })
  const realtime = server.app.container.use('realtime')
  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. To test the cache after an event, call the subscription's effects directly:

// 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() and push events from the test.

Extend

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

// 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:

// app/components/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()
})

A real transport also resubscribes after reconnecting. See SocketTransport in the API reference.

Write a server adapter

A RealtimeAdapter carries emits between instances; implement it to use another bus. subscribe(topic, handler) starts delivering a topic's messages and returns the function that stops it, and publish(topic, payload) sends to every instance and returns how many received it. This one wraps Postgres LISTEN/NOTIFY with pg:

// server/realtime/postgres_adapter.ts
import { createHash } from 'node:crypto'
import type { RealtimeAdapter } from '@kitr/realtime/server'
import type { Client } from 'pg'

export function createPostgresAdapter(listen: Client, notify: Client): RealtimeAdapter {
  const handlers = new Map<string, Set<(payload: string) => void>>()
  // Postgres channel names are short identifiers: hash each topic into one.
  const channel = (topic: string) => `rt_${createHash('sha1').update(topic).digest('hex')}`
  const topics = new Map<string, string>()
  listen.on('notification', (msg) => {
    const topic = topics.get(msg.channel)
    for (const handler of handlers.get(topic ?? '') ?? []) handler(msg.payload ?? '')
  })
  return {
    name: 'postgres',
    async subscribe(topic, handler) {
      if (!handlers.has(topic)) {
        handlers.set(topic, new Set())
        topics.set(channel(topic), topic)
        await listen.query(`LISTEN ${channel(topic)}`)
      }
      handlers.get(topic)!.add(handler)
      return () => {
        handlers.get(topic)?.delete(handler)
        if (handlers.get(topic)?.size) return
        handlers.delete(topic)
        void listen.query(`UNLISTEN ${channel(topic)}`)
      }
    },
    async publish(topic, payload) {
      await notify.query('SELECT pg_notify($1, $2)', [channel(topic), payload])
      // Postgres does not report receivers; 1 keeps SSE control forwarding working.
      return 1
    },
  }
}

Pass it as adapter on .ws() or .sse(), as with Redis. An adapter that cannot count receivers should return 1: the SSE server answers 410 only when publish returns 0.

Next steps