Realtime

How It Works

Realtime has two halves, and both have to be switched on:

1. On the server, a realtime adapter passed to createNotifyKit({ realtime }). Every inbox change (a new item, a read, an archive, a delete) is published to it, keyed by recipient, tenant, and workspace.
2. In the browser, realtime: true on the React provider or client. The hooks then open GET /inbox/stream, a server-sent event stream that the handler subscribes to on behalf of the signed-in user.

Without the server half the stream route returns 404; without the browser half the hooks never open it. Either way everything else still works, with the inbox loading on mount and on refresh().

import { createNotifyKit, memoryRealtimeAdapter } from "@notifykitjs/core"

export const notify = createNotifyKit({
	// ...
	realtime: memoryRealtimeAdapter(),
})

<NotifyKitProvider options={{ baseUrl: "/api/notifykit", realtime: true }}>

Adapters

Swapping one for another is a change to the realtime option only; the browser code stays the same. The WebSocket adapter keeps its subscribers in memory like the memory adapter, so it does not carry events between servers either.

Postgres LISTEN/NOTIFY

npm install @notifykitjs/realtime-pg

The adapter does not depend on a Postgres driver; you give it three functions. LISTEN needs one connection of its own, held open, rather than one borrowed from the pool per query. With pg:

import { Pool } from "pg"
import { pgRealtimeAdapter } from "@notifykitjs/realtime-pg"

const pool = new Pool({ connectionString: process.env.DATABASE_URL })
const listener = await pool.connect()

export const realtime = pgRealtimeAdapter({
	connection: {
		listen: async (channel, handler) => {
			listener.on("notification", (msg) => {
				if (msg.channel === channel && msg.payload) handler(msg.payload)
			})
			await listener.query(`LISTEN ${channel}`)
		},
		unlisten: async (channel) => { await listener.query(`UNLISTEN ${channel}`) },
		notify: async (channel, payload) => {
			await pool.query("SELECT pg_notify($1, $2)", [channel, payload])
		},
	},
	onError: (err) => console.error("[realtime]", err),
})

await realtime.start()

Call start() before sending and stop() on shutdown. Postgres limits a notification to about 8 KB; a larger event is replaced by an inbox.refetch, which makes clients reload the list.

WebSocket

npm install @notifykitjs/realtime-ws

The adapter does not run a server. You accept connections with your WebSocket library and hand each one to it; it authenticates the upgrade request and routes events to the right sockets.

import { WebSocketServer } from "ws"
import { webSocketRealtimeAdapter } from "@notifykitjs/realtime-ws"

export const realtime = webSocketRealtimeAdapter({
	authenticate: async (request) => {
		const token = new URL(request.url).searchParams.get("token")
		const session = await verifyToken(token)
		if (!session) return null
		return { recipientId: session.userId, tenantId: session.orgId }
	},
	allowedOrigins: ["https://app.example.com"],
})

new WebSocketServer({ server }).on("connection", async (ws, req) => {
	const request = new Request(`https://${req.headers.host}${req.url}`, {
		headers: { origin: req.headers.origin ?? "" },
	})
	if (!(await realtime.handleUpgrade(request, ws))) {
		return ws.close(4001, "Unauthorized")
	}
	ws.on("message", (data) => realtime.handleMessage(ws, data.toString()))
	ws.on("close", () => realtime.handleClose(ws))
})

Clients receive each event as a JSON message. The React client uses the SSE stream, not WebSockets, so this adapter is for clients you write yourself, such as a mobile app.

Events

Events only reach the recipient they are for, in the same tenant and workspace. A stream subscribed for one tenant never sees another tenant’s events for the same user id.

The Stream

• The handler sends a heartbeat comment every 30 seconds. Set proxy and load balancer idle timeouts above that.
• Each stream closes after 30 minutes and the client reconnects, which keeps long-lived connections fresh.
• One user can hold ten streams at once per server; the eleventh gets a 429.
• If a client falls behind, the events it missed are replaced by one inbox.refetch.

The React client handles the rest:

• It reads the stream with fetch, not EventSource, so your headers and credentials options apply to it.
• When the stream drops it reconnects after 1 second, doubling to at most 30, with jitter, for up to 20 tries in a row. After every reconnect it reloads the inbox, so nothing sent during the gap is missed.
• A 401 or 403 stops it for good, until the page reloads.
• Several hooks share one stream, which closes when the last of them unmounts.

realtimeStatus on useInbox() reports "connecting", "connected", or "disconnected"; onRealtimeError on the provider receives each failure.

Serverless and Polling

A stream lives only as long as the function serving it. On Vercel it ends at the function’s maximum duration and the client reconnects, which works but costs a reconnect and a reload each time. Where that is too often, poll instead:

const { items } = useInbox({ pollInterval: 15_000 })
const { unreadCount } = useUnreadCount({ pollInterval: 15_000 })

With pollInterval set, a hook polls and does not open the stream. Ten to thirty seconds suits most apps; each poll is one request per open tab. The memory adapter is also per process, so on serverless an event published in one invocation never reaches a stream held by another; use Postgres there.

Checking It Works

curl -N -H "Cookie: session=..." http://localhost:3000/api/notifykit/inbox/stream

You should see : connected straight away and : heartbeat every 30 seconds. Send the user a notification and an event: inbox.created line follows. If curl sees events but the page does not update, the problem is in the browser (no realtime: true, or no provider); if curl sees nothing, it is on the server.

Testing

With the memory adapter, events are delivered before send() resolves, so tests can subscribe, send, and assert:

const realtime = memoryRealtimeAdapter()
const notify = createNotifyKit({ /* ... */ realtime })

const events: RealtimeEvent[] = []
realtime.subscribe("alice", { tenantId: "org_a" }, (e) => events.push(e))

await notify.send({
	recipientId: "alice",
	tenantId: "org_a",
	notificationId: "comment_mentioned",
	payload,
})

expect(events[0]).toMatchObject({
	type: "inbox.created",
	item: { recipientId: "alice" },
})