diff --git a/apps/web/content/docs/content-engine/live-editing-api.mdx b/apps/web/content/docs/content-engine/live-editing-api.mdx index 75a1cd921..34441af01 100644 --- a/apps/web/content/docs/content-engine/live-editing-api.mdx +++ b/apps/web/content/docs/content-engine/live-editing-api.mdx @@ -8,12 +8,14 @@ The AdminCP drives [live editing](/docs/content-engine/live-editing) for you. Th ## Deployment requirements -Live editing runs on two channels. Field locks and the shared draft are plain HTTP routes, so they work everywhere. Presence and rich text co-editing travel over the [WebSocket](/docs/dev/websocket) at `/api/ws`, which only a long-lived Node API server provides, such as `apps/api` or a [self-hosted](/docs/dev/deployments/self-hosted) install. +Live editing runs on two channels. Field locks and the shared draft are plain HTTP routes, so they work everywhere. Presence and rich text co-editing travel over the [WebSocket](/docs/dev/websocket) at `/api/ws`, which a standalone API server or a single app with the [Nitro WebSocket route](/docs/dev/websocket/setup) provides, on [Vercel](/docs/dev/deployments/cloud/vercel) too. -| Deployment | Field locks and autosave | Presence | Rich text | -| ------------------------------------------------------------------ | ---------------------------- | -------- | ----------------------------------------------- | -| Long-lived Node API server with the WebSocket | Live, pushed over the socket | Yes | Co-edited | -| No WebSocket, such as [Vercel](/docs/dev/deployments/cloud/vercel) | Polled over HTTP every 10 s | No | One editor at a time, locked like a plain field | +| Deployment | Field locks and autosave | Presence | Rich text | +| --------------------------------------------- | ---------------------------- | -------- | ----------------------------------------------- | +| WebSocket at `/api/ws` | Live, pushed over the socket | Yes | Co-edited | +| No WebSocket, or while the socket reconnects | Polled over HTTP every 10 s | No | One editor at a time, locked like a plain field | + +After a reconnect each editor joins the room again, reloads the locks and the draft, and reopens its rich text documents, which fetches the edits it missed. A form without a socket polls `GET /{id}/locks` and `GET /{id}/draft` every 10 seconds, on top of lock renewals. Every request counts against the [rate limiter](/docs/dev/advanced/rate-limiter), which is keyed by IP address: the polls against the read budget (300 a minute), lock renewals and draft saves against the write budget (60 a minute). A newsroom behind one office IP can reach the write budget. A refused lock request leaves the field without autosave, so raise `rateLimiter.write.points` if that is you. diff --git a/apps/web/content/docs/dev/advanced/redis.mdx b/apps/web/content/docs/dev/advanced/redis.mdx index a1388cd73..530b75b82 100644 --- a/apps/web/content/docs/dev/advanced/redis.mdx +++ b/apps/web/content/docs/dev/advanced/redis.mdx @@ -38,7 +38,7 @@ export const vitNodeApiConfig = buildApiConfig({ | :--- | :--- | :--- | | **API Domain Cache** | No-op (always executes DB query) | Shared key-value store via `c.get("cache")` | | **Rate Limiter** | In-memory counters per instance | Distributed sliding-window across cluster | -| **WebSocket Realtime** | Single instance only | Cross-instance broadcasts via Redis Pub/Sub | +| **WebSocket Realtime** | Single instance only | Cross-instance delivery via Redis Pub/Sub; required on serverless hosts ([details](/docs/dev/websocket/reconnects#delivery-across-instances)) | | **Sessions** | Database queries on every request | Instant cache hits with DB fallback | --- diff --git a/apps/web/content/docs/dev/configuration.mdx b/apps/web/content/docs/dev/configuration.mdx index c0c4c161f..5d74ab7f0 100644 --- a/apps/web/content/docs/dev/configuration.mdx +++ b/apps/web/content/docs/dev/configuration.mdx @@ -344,8 +344,9 @@ the redirect and once for the page. ## `vite.config.ts` One VitNode plugin, which is the environment handling, the dev server's -dependency pre-bundling, the SSR externals, the Tailwind source list and the -plugin route generator in the order they have to run: +dependency pre-bundling, the SSR externals, the Tailwind source list, the +plugin route generator and Nitro, the server your app runs on, in the order they +have to run: ```ts title="apps/web/vite.config.ts" import { vitnode } from '@vitnode/core/framework/vite' @@ -353,7 +354,6 @@ import { vitnode } from '@vitnode/core/framework/vite' export default defineConfig({ plugins: [ vitnode({ appRoot: import.meta.dirname }), - nitro(), tailwindcss(), tanstackStart(), viteReact(), @@ -361,12 +361,32 @@ export default defineConfig({ }) ``` +`vitnode()` already adds `nitro()`, so don't add it again. Nitro's own options go +in `nitro` - a deployment `preset`, `routeRules`, extra `handlers`, anything from +[Nitro's config](https://nitro.build/config): + +```ts title="apps/web/vite.config.ts" +vitnode({ + appRoot: import.meta.dirname, + nitro: { + preset: 'vercel', + rollupConfig: { external: [/^@sentry\//] }, + routeRules: { '/legacy/**': { redirect: '/' } }, + }, +}) +``` + +VitNode adds its own Nitro settings on top of yours: the SSR externals in a +build, the development error page in `vite dev`, and the `/api/ws` route with +`webSocket: true`. Lists such as `handlers` are merged, so yours and VitNode's +both run. + The Node server from `pnpm start` sends static files and HTML uncompressed. Put a reverse proxy, a CDN or a platform such as Vercel in front of it and let that layer compress responses. Nitro's `compressPublicAssets` can write a `.br` and a `.gz` twin next to every -built asset instead, but it stays off unless you pass it to `nitro()` yourself. +built asset instead, but it stays off unless you turn it on in `nitro` yourself. It compresses every file in the output at once, and on a 2-core, 8 GB build machine that is enough to run out of memory. The twins only help when Nitro itself serves the files: behind a CDN they are never read. diff --git a/apps/web/content/docs/dev/deployments/cloud/vercel.mdx b/apps/web/content/docs/dev/deployments/cloud/vercel.mdx index 33ab51622..ae859bffb 100644 --- a/apps/web/content/docs/dev/deployments/cloud/vercel.mdx +++ b/apps/web/content/docs/dev/deployments/cloud/vercel.mdx @@ -10,9 +10,10 @@ A VitNode app builds with Nitro, and Nitro knows Vercel. Push the repository, point a project at it, and the site plus the Hono API mounted at `/api/*` come up as one serverless function - no adapter, no `vercel.json`. -Serverless is the catch, not the build. Three things VitNode can do need a +Serverless is the catch, not the build. Two things VitNode can do need a process that is still alive between requests, and Vercel does not have one - and -nothing there will run your migrations either. +nothing there will run your migrations either. WebSockets are not one of them: +they run on Vercel Functions, with a time limit per connection. ## What works, and what does not @@ -22,7 +23,7 @@ nothing there will run your migrations either. | Postgres, [Redis](/docs/dev/advanced/redis), email, AI | ✅ Works | All reached over the network. Bring managed ones. | | [S3 / R2](/docs/dev/storage/s3-r2) or Supabase storage | ✅ Works | Uploads go straight to the bucket. | | [Local file storage](/docs/dev/storage/local) | ❌ Not durable | The adapter writes to `public/uploads` on a filesystem that disappears. | -| [WebSockets](/docs/dev/websocket) | ❌ Unsupported | The `/ws` upgrade needs a long-lived `serve()` with a WebSocket server. | +| [WebSockets](/docs/dev/websocket) | ✅ Works | Nitro serves `/api/ws`. See [WebSockets](#websockets) for the limits. | | [In-process cron](/docs/dev/cron/node-cron) | ❌ Unsupported | The adapter runs a `*/1 * * * *` timer inside the process. | | Migrating on deploy | ⚠️ Your job | Nothing on Vercel runs `db:migrate` for you. See [Migrations](#migrations). | @@ -135,11 +136,35 @@ empty database becomes the **Administrator** - so do it before you tell anybody the URL. Then sign in at `/admin` and open **Core → System → Integrations** to see which services actually came up. -{/* Image prompt: A screenshot of the VitNode AdminCP Integrations board at /admin/core/system/integrations on a production deployment - nine service cards, Redis and Storage showing "Active" badges, WebSocket and Cron showing "Inactive". Dark theme, 1440x900. */} +{/* Image prompt: A screenshot of the VitNode AdminCP Integrations board at /admin/core/system/integrations on a production deployment - nine service cards, Redis, Storage and WebSocket showing "Active" badges, Cron showing "Inactive". Dark theme, 1440x900. */} +## WebSockets + +A single app serves `/api/ws` from the same deployment as the site, through +Nitro's Vercel preset - no separate server and no realtime service. Set it up as +in [Set up the endpoint](/docs/dev/websocket/setup); new single apps already +are. Vercel runs WebSocket connections on +[Fluid compute](https://vercel.com/docs/fluid-compute), the default for new +projects, so check it is enabled on older ones. + +Three things behave differently from a server that stays up: + +- **Connections end.** Vercel closes a socket when its function reaches the + maximum duration. The browser reconnects with backoff, rejoins its rooms and + reloads missed state - see + [Reconnects and missed messages](/docs/dev/websocket/reconnects). Limits + depend on your plan; check Vercel's current + [function limits](https://vercel.com/docs/functions/limitations). +- **Instances differ.** The request that sends a notification and the socket + that should receive it often run on different instances of the same + deployment. Set `REDIS_URL`, or messages reach only sockets on the sending + instance. +- **Nothing is replayed.** A socket that is reconnecting misses what was sent + meanwhile. VitNode's own features reload it from the database; yours should too. + ## Migrations `start` never migrates, and on Vercel there is no `start` at all - your code is @@ -196,10 +221,9 @@ deployment, and there are exactly two honest ways to do it: controls - see [Swagger](/docs/dev/swagger). - - Run the app on a server instead. Long-lived WebSocket connections, an - in-process scheduler and local disk are all ordinary there - see - [Self-hosted](/docs/dev/deployments/self-hosted). + + Run the app on a server instead. An in-process scheduler and local disk are + both ordinary there - see [Self-hosted](/docs/dev/deployments/self-hosted). ## Next diff --git a/apps/web/content/docs/dev/deployments/prerendering.mdx b/apps/web/content/docs/dev/deployments/prerendering.mdx index d70c7fca4..1208d73da 100644 --- a/apps/web/content/docs/dev/deployments/prerendering.mdx +++ b/apps/web/content/docs/dev/deployments/prerendering.mdx @@ -30,11 +30,10 @@ A route with a parameter, such as `/solutions/$slug`, is never found on its own. ### Hand the list to Nitro -Pass the paths through `prerenderRoutes` from `@vitnode/core/framework/vite` and give the result to the `prerender.routes` option of `nitro()`: +Pass the paths through `prerenderRoutes` from `@vitnode/core/framework/vite` and give the result to Nitro's `prerender.routes`, through the `nitro` option of `vitnode()`: ```ts title="vite.config.ts" import { prerenderRoutes, vitnode } from '@vitnode/core/framework/vite' // [!code highlight] -import { nitro } from 'nitro/vite' import { defineConfig } from 'vite' import { PRERENDERED_PATHS } from './src/prerender.ts' // [!code highlight] @@ -42,14 +41,16 @@ import { vitNodeConfig } from './src/vitnode.config.ts' // [!code highlight] export default defineConfig({ plugins: [ - vitnode({ appRoot: import.meta.dirname }), - nitro({ - // [!code highlight:6] - prerender: { - routes: prerenderRoutes({ - i18n: vitNodeConfig.i18n, - paths: PRERENDERED_PATHS, - }), + vitnode({ + appRoot: import.meta.dirname, + // [!code highlight:8] + nitro: { + prerender: { + routes: prerenderRoutes({ + i18n: vitNodeConfig.i18n, + paths: PRERENDERED_PATHS, + }), + }, }, }), // tailwindcss(), tanstackStart(), viteReact() as before diff --git a/apps/web/content/docs/dev/websocket.mdx b/apps/web/content/docs/dev/websocket/index.mdx similarity index 79% rename from apps/web/content/docs/dev/websocket.mdx rename to apps/web/content/docs/dev/websocket/index.mdx index 5b7847a9b..feaaedd53 100644 --- a/apps/web/content/docs/dev/websocket.mdx +++ b/apps/web/content/docs/dev/websocket/index.mdx @@ -6,7 +6,7 @@ icon: Radio import { Tab, Tabs } from "fumadocs-ui/components/tabs" -VitNode provides a unified, multiplexed WebSocket connection at `/api/ws`. Browsers share a single connection across tabs via a Web Lock leader election, authenticated by session cookies. +VitNode provides a unified, multiplexed WebSocket connection at `/api/ws`. Browsers share a single connection across tabs via a Web Lock leader election, authenticated by session cookies. A single app serves it from the same deployment as the site and the API, Vercel included - see [Set up the endpoint](/docs/dev/websocket/setup). ## Quick start @@ -158,6 +158,23 @@ c.get("realtime").toRoom(`chat:${roomId}`, chatChannel, { Each socket in the room receives the payload once, however many of its tabs joined. Put the room or record in the payload so every tab can ignore what is not its own. +### Join again after a reconnect + +A new connection starts with no rooms on the server. `useVitNodeWebSocket` returns a `connectionId` that changes every time the shared connection opens, in every tab, so join from an effect that depends on it. `roomsChannel` is `createWebSocketChannel({ pluginId: "@acme/chat", module: "rooms", id: "presence" })`, the channel of `roomsSocket` above: + +```tsx title="plugins/chat/src/views/chat-room.tsx" +const [clientId] = React.useState(() => crypto.randomUUID()) +const { connectionId, send } = useVitNodeWebSocket(roomsChannel) + +// [!code ++:4] +React.useEffect(() => { + if (!connectionId) return + send({ clientId, roomId, type: "join" }) +}, [clientId, connectionId, roomId, send]) +``` + +[Reconnects and missed messages](/docs/dev/websocket/reconnects) covers reloading what the tab missed while it was away. + ### Clean up when a socket closes A socket that closes leaves all of its rooms. Use `onConnectionClose` to update your own state, such as an "online now" list: @@ -194,13 +211,23 @@ Without Redis, all of this works on a single instance with no extra setup: `toRo - **Tab Leader Election**: When a user opens multiple tabs, a Web Lock elects one leader tab to hold the socket, distributing messages across tabs via `BroadcastChannel`. - **Automatic Reconnect**: A dropped connection retries after about a second, then backs off (with a little jitter) up to 30 seconds, so a restarted server is not stampeded by every open tab. - **Offline aware**: When the browser goes offline, the socket closes and stops retrying. It reconnects the moment the network comes back, or when a tab becomes visible again after sleep, without waiting for the backoff. -- **Heartbeat, both ways**: The bundled API server pings every socket every 30 seconds and closes the ones that stopped answering, so vanished clients do not linger. The browser sends its own ping every 25 seconds and reconnects when no reply arrives within 10 seconds, so a half-dead connection after a Wi-Fi drop never pretends to be online. -- **No replay**: Messages sent while a socket is down are not queued. A feature that must not miss updates sends its current state when the socket opens, like the notification count does. +- **Heartbeat, both ways**: The server pings every socket every 30 seconds and closes the ones that stopped answering, so vanished clients do not linger (the standalone API's `ws` server and Nitro's crossws both do). The browser sends its own ping every 25 seconds and reconnects when no reply arrives within 10 seconds, so a half-dead connection after a Wi-Fi drop never pretends to be online. +- **No replay**: Messages sent while a socket is down are not queued. A feature that must not miss updates reloads its state when the socket opens again - see [Reconnects and missed messages](/docs/dev/websocket/reconnects). - **Multiple Instances**: With `REDIS_URL` set, `sendToUser`, `broadcast` and `toRoom` are relayed to every API instance over Redis pub/sub. Without Redis they only reach clients connected to the same instance. ## Learn More + + appendMessage(message), +}) + +React.useEffect(() => { + if (!connectionId) return + + send({ clientId, roomId, type: "join" }) // the server forgot the room + void queryClient.invalidateQueries({ queryKey: ["chat", roomId] }) // reload what was missed +}, [clientId, connectionId, queryClient, roomId, send]) +``` + +Messages sent before the socket is open are queued and sent once it opens, so the join is never lost. Do not resend writes from that effect: a mutation that went over HTTP already happened, and sending it again would do it twice. + +VitNode's own features follow the same rule: + +- **Notifications**: the server sends the user's unread count and revision as soon as a socket opens. Every tab compares it with what it has and refetches the inbox when it changed. While no socket is open, the bell polls every 60 seconds. +- **Live editing**: each tab joins its record's room on every new connection, then reloads the field locks and the shared draft, and reopens its rich text documents so the server sends the edits it missed. Without a socket the form polls over HTTP. + +## Delivery across instances + +`c.get("realtime")` delivers to the sockets on the instance that called it, then publishes the message to Redis. Every other instance delivers it to its own matching sockets and ignores its own echo, so nobody gets a message twice. An instance accepts sockets only once it is subscribed. + +| Setup | Who receives a message | +| ------------------------- | -------------------------------------------------------------- | +| No `REDIS_URL` | Sockets on the same instance only. Fine for one server. | +| `REDIS_URL`, one instance | Same as above, plus a ready check before sockets are accepted. | +| `REDIS_URL`, many | Matching sockets on every instance. | + +On serverless hosts such as Vercel the HTTP request that sends a message and the socket that should receive it often run on different instances, even within one deployment, so set `REDIS_URL` there. A publish started during a request is kept alive with the request's `waitUntil`, so the instance is not frozen before Redis has it. + +When the Redis subscription drops and comes back, the instance closes its sockets with code `1012`. The clients reconnect and run the restore above, which picks up whatever was published while the instance could not hear it. While Redis is down, new handshakes on that instance answer `503` and the client keeps retrying. + +## Related + + + + + diff --git a/apps/web/content/docs/dev/websocket/setup.mdx b/apps/web/content/docs/dev/websocket/setup.mdx new file mode 100644 index 000000000..161bbc6b1 --- /dev/null +++ b/apps/web/content/docs/dev/websocket/setup.mdx @@ -0,0 +1,125 @@ +--- +title: Set up the WebSocket endpoint +description: Serve VitNode's /api/ws WebSocket from a single TanStack Start app with Nitro, Vercel included, or from a standalone Node API server. +icon: Plug +--- + +VitNode's WebSocket lives at `/api/ws`. A single app serves it from the same deployment as the site and the HTTP API, through Nitro. A standalone API serves it from its own Node server. Both run the same handshake, protocol and handlers, so plugins and `useVitNodeWebSocket` work the same either way. + +New single apps from `create-vitnode-app` already include everything on this page. Follow it to add WebSockets to an older single app. + +## Single app + +The browser upgrades `/api/ws`. Nitro owns that route, so the upgrade never reaches the TanStack Start catch-all, while every other `/api/*` request still does. Nitro then sends the handshake through the embedded Hono API as an ordinary `GET /api/ws`, so the Origin check, the rate limiter and the session cookie apply before the socket opens. + + + + +### Enable WebSockets in the Vite config + +Tell VitNode this app serves its own socket: + +```ts title="vite.config.ts" +plugins: [ + vitnode({ appRoot: import.meta.dirname, webSocket: true }), // [!code highlight] + devtools(), + tailwindcss(), + tanstackStart(), + viteReact(), +], +``` + +That one flag turns on Nitro's WebSocket support, mounts VitNode's Nitro route at `/api/ws` and lets the browser open the socket. Without it the browser never connects in an app that embeds the API. + +The route ships with `@vitnode/core`, so there is no file to copy. It runs in Nitro's bundle, loads your server bundle on the first upgrade a fresh instance gets, and hands every handshake to the API your server bundle registered with `configureInProcessApi`. If that bundle cannot load, the handshake gets `503` and the client retries. + + +Older single apps mounted their own `src/server/websocket.ts` through `nitro({ handlers })`. Delete that file and the `nitro()` call when you switch to `webSocket: true`, or two handlers fight over `/api/ws`. + + + + + +### Mount the handshake on the embedded API + +Add the `/ws` route after `VitNodeAPI`, so the API's middleware runs first: + +```ts title="src/server/vitnode-api.server.ts" +import { websocketOriginMiddleware } from "@vitnode/core/api/middlewares/websocket-origin.middleware" // [!code ++] +import { upgradeWebSocket } from "@vitnode/core/ws/crossws.server" // [!code ++] +import { handleVitNodeWebSocket } from "@vitnode/core/ws/handle" // [!code ++] + +const createVitNodeApi = () => { + const app = new OpenAPIHono().basePath("/api") + + VitNodeAPI({ app, vitNodeApiConfig }) + + // [!code ++:5] + app.get( + "/ws", + websocketOriginMiddleware(), + upgradeWebSocket(handleVitNodeWebSocket()), + ) + + return app +} +``` + +`upgradeWebSocket` from `@vitnode/core/ws/crossws.server` does not upgrade anything itself. It hands the connection's handlers back to the Nitro route, which opens the socket. A plain `GET /api/ws` gets `426 Upgrade Required`. + + + + +## Standalone API + +A separate API server runs on `@hono/node-server` and upgrades with its own helper and a `ws` server. Mount the route after `VitNodeAPI`, and pass the web app's origin when it is not `VITNODE_WEB_URL`: + +```ts title="apps/api/src/index.ts" +import { serve, upgradeWebSocket } from "@hono/node-server" +import { websocketOriginMiddleware } from "@vitnode/core/api/middlewares/websocket-origin.middleware" +import { handleVitNodeWebSocket } from "@vitnode/core/ws/handle" +import { WebSocketServer } from "ws" + +app.get( + "/ws", + websocketOriginMiddleware({ origin: [webOrigin] }), + upgradeWebSocket(handleVitNodeWebSocket()), +) + +serve({ + fetch: app.fetch, + port: Number(process.env.PORT ?? 8000), + websocket: { server: new WebSocketServer({ noServer: true }) }, +}) +``` + +A web app pointed at that API with `VITNODE_API_URL` connects to it on its own; it needs no `webSocket` option and no Nitro route. + +## Who may connect + +Both setups apply the same rules before a socket opens: + +- **Origin**: the configured web and API origins, any extra `origin` you pass, and the handshake's own origin (a page on the same host, such as a preview deployment). A handshake without an `Origin` header is not a browser and is allowed. Anything else gets `403`. +- **Session**: the session cookie decides the user. Guests connect too, as `null`; a handler that needs a user checks `c.get("user")` itself, and live editing checks the AdminCP session on every message. +- **Redis**: with `REDIS_URL` set, the instance answers `503` until it is subscribed to the realtime channel, so it never accepts a socket that other instances cannot reach. The client retries. + +A rejected handshake registers nothing. Identity is fixed for the life of a socket: after sign-in or sign-out the client reconnects on its own, but a session revoked from another device keeps its open socket until that socket closes. + +## Check the result + +Run the app, open the site, and look at the browser's network panel: `/api/ws` shows `101 Switching Protocols`, followed by `vitnode_ping` and `vitnode_pong` frames every 25 seconds. **AdminCP → Core → System → Integrations** shows WebSocket as active. + +## Related + + + + + diff --git a/apps/web/src/server/vitnode-api.server.ts b/apps/web/src/server/vitnode-api.server.ts index 3d36b7b78..e73bb39a5 100644 --- a/apps/web/src/server/vitnode-api.server.ts +++ b/apps/web/src/server/vitnode-api.server.ts @@ -1,7 +1,10 @@ import '@tanstack/react-start/server-only' import { OpenAPIHono } from '@hono/zod-openapi' import { VitNodeAPI } from '@vitnode/core/api/config' +import { websocketOriginMiddleware } from '@vitnode/core/api/middlewares/websocket-origin.middleware' import { configureInProcessApi } from '@vitnode/core/lib/fetcher/in-process-api.server' +import { upgradeWebSocket } from '@vitnode/core/ws/crossws.server' +import { handleVitNodeWebSocket } from '@vitnode/core/ws/handle' import { createApiBridge } from '@/server/api-bridge' import { vitNodeApiConfig } from '@/vitnode.api.config' @@ -14,6 +17,16 @@ const createVitNodeApi = () => { VitNodeAPI({ app, vitNodeApiConfig }) + // The handshake half of `/api/ws`: Nitro's route (`./websocket.ts`) sends + // each upgrade through here, so it passes the Origin check and the API's + // middleware before the socket opens. A handshake is a cookie-authenticated + // GET that neither CSRF nor the same-origin policy covers - see the middleware. + app.get( + '/ws', + websocketOriginMiddleware(), + upgradeWebSocket(handleVitNodeWebSocket()), + ) + return app } diff --git a/apps/web/vite.config.ts b/apps/web/vite.config.ts index 9a1c42999..67f27e243 100644 --- a/apps/web/vite.config.ts +++ b/apps/web/vite.config.ts @@ -4,7 +4,6 @@ import { tanstackStart } from '@tanstack/react-start/plugin/vite' import viteReact from '@vitejs/plugin-react' import { prerenderRoutes, vitnode } from '@vitnode/core/framework/vite' import { fumadocsMdx } from 'fumadocs-mdx/vite' -import { nitro } from 'nitro/vite' import { defineConfig } from 'vite' import { docsPrerenderPaths } from './src/docs/prerender.ts' @@ -18,17 +17,20 @@ const config = defineConfig({ }, server: { strictPort: true }, plugins: [ - vitnode({ appRoot: import.meta.dirname }), - fumadocsMdx({ index: false }), - devtools(), - nitro({ - prerender: { - routes: prerenderRoutes({ - i18n: vitNodeConfig.i18n, - paths: [...PRERENDERED_PATHS, ...docsPrerenderPaths()], - }), + vitnode({ + appRoot: import.meta.dirname, + nitro: { + prerender: { + routes: prerenderRoutes({ + i18n: vitNodeConfig.i18n, + paths: [...PRERENDERED_PATHS, ...docsPrerenderPaths()], + }), + }, }, + webSocket: true, }), + fumadocsMdx({ index: false }), + devtools(), tailwindcss(), tanstackStart(), viteReact(), diff --git a/packages/create-vitnode-app/copy-of-vitnode-app/api-single-app/src/server/vitnode-api.server.ts b/packages/create-vitnode-app/copy-of-vitnode-app/api-single-app/src/server/vitnode-api.server.ts index 752ea2f94..bf1a17eed 100644 --- a/packages/create-vitnode-app/copy-of-vitnode-app/api-single-app/src/server/vitnode-api.server.ts +++ b/packages/create-vitnode-app/copy-of-vitnode-app/api-single-app/src/server/vitnode-api.server.ts @@ -1,7 +1,10 @@ import "@tanstack/react-start/server-only"; import { OpenAPIHono } from "@hono/zod-openapi"; import { VitNodeAPI } from "@vitnode/core/api/config"; +import { websocketOriginMiddleware } from "@vitnode/core/api/middlewares/websocket-origin.middleware"; import { configureInProcessApi } from "@vitnode/core/lib/fetcher/in-process-api.server"; +import { upgradeWebSocket } from "@vitnode/core/ws/crossws.server"; +import { handleVitNodeWebSocket } from "@vitnode/core/ws/handle"; import { createApiBridge } from "@/server/api-bridge"; import { vitNodeApiConfig } from "@/vitnode.api.config"; @@ -14,6 +17,16 @@ const createVitNodeApi = () => { VitNodeAPI({ app, vitNodeApiConfig }); + // The handshake half of `/api/ws`: Nitro's route (`./websocket.ts`) sends + // each upgrade through here, so it passes the Origin check and the API's + // middleware before the socket opens. A handshake is a cookie-authenticated + // GET that neither CSRF nor the same-origin policy covers - see the middleware. + app.get( + "/ws", + websocketOriginMiddleware(), + upgradeWebSocket(handleVitNodeWebSocket()), + ); + return app; }; diff --git a/packages/create-vitnode-app/copy-of-vitnode-app/api-single-app/vite.config.ts b/packages/create-vitnode-app/copy-of-vitnode-app/api-single-app/vite.config.ts new file mode 100644 index 000000000..8dc2b5cad --- /dev/null +++ b/packages/create-vitnode-app/copy-of-vitnode-app/api-single-app/vite.config.ts @@ -0,0 +1,24 @@ +import tailwindcss from "@tailwindcss/vite"; +import { devtools } from "@tanstack/devtools-vite"; +import { tanstackStart } from "@tanstack/react-start/plugin/vite"; +import viteReact from "@vitejs/plugin-react"; +import { vitnode } from "@vitnode/core/framework/vite"; +import { defineConfig } from "vite"; + +const config = defineConfig({ + resolve: { tsconfigPaths: true }, + server: { strictPort: true }, + plugins: [ + vitnode({ + appRoot: import.meta.dirname, + nitro: { rollupConfig: { external: [/^@sentry\//] } }, + webSocket: true, + }), + devtools(), + tailwindcss(), + tanstackStart(), + viteReact(), + ], +}); + +export default config; diff --git a/packages/create-vitnode-app/copy-of-vitnode-app/root/vite.config.ts b/packages/create-vitnode-app/copy-of-vitnode-app/root/vite.config.ts index a4312ebdd..34b1e8bd7 100644 --- a/packages/create-vitnode-app/copy-of-vitnode-app/root/vite.config.ts +++ b/packages/create-vitnode-app/copy-of-vitnode-app/root/vite.config.ts @@ -3,16 +3,17 @@ import { devtools } from "@tanstack/devtools-vite"; import { tanstackStart } from "@tanstack/react-start/plugin/vite"; import viteReact from "@vitejs/plugin-react"; import { vitnode } from "@vitnode/core/framework/vite"; -import { nitro } from "nitro/vite"; import { defineConfig } from "vite"; const config = defineConfig({ resolve: { tsconfigPaths: true }, server: { strictPort: true }, plugins: [ - vitnode({ appRoot: import.meta.dirname }), + vitnode({ + appRoot: import.meta.dirname, + nitro: { rollupConfig: { external: [/^@sentry\//] } }, + }), devtools(), - nitro({ rollupConfig: { external: [/^@sentry\//] } }), tailwindcss(), tanstackStart(), viteReact(), diff --git a/packages/vitnode/package.json b/packages/vitnode/package.json index 4fd6f8bfd..23d0f14fb 100644 --- a/packages/vitnode/package.json +++ b/packages/vitnode/package.json @@ -43,6 +43,7 @@ "drizzle-orm": "^1.0.0-rc.4", "hono": "^4.9.x", "motion": "^14.x.x", + "nitro": "^3.0.260903-beta", "react": "^19.2.x", "react-dom": "^19.2.x", "tsdown": "^0.23.0", @@ -76,6 +77,7 @@ "jiti": "^2.7.0", "jsdom": "^30.1.2", "lucide-react": "^1.53.0", + "nitro": "3.0.260903-beta", "oxlint": "^1.87.0", "oxlint-tsgolint": "^7.0.2003", "react": "^19.3.0", @@ -313,6 +315,9 @@ "@tanstack/react-start": { "optional": true }, + "nitro": { + "optional": true + }, "tsdown": { "optional": true }, diff --git a/packages/vitnode/src/api/middlewares/global.middleware.ts b/packages/vitnode/src/api/middlewares/global.middleware.ts index 251af4377..347cbb408 100644 --- a/packages/vitnode/src/api/middlewares/global.middleware.ts +++ b/packages/vitnode/src/api/middlewares/global.middleware.ts @@ -14,6 +14,7 @@ import type { VitNodeRealtime } from "@/ws/registry"; import { LocalEventsAdapter } from "@/api/adapters/events/local"; import { PostgresSearchAdapter } from "@/api/adapters/search/postgres"; +import { runAfterResponse } from "@/api/lib/after-response"; import { DEFAULT_SESSION_COOKIE_NAME } from "@/api/lib/auth-cookie"; import { CacheModel } from "@/api/lib/cache"; import { MemoryCacheStore } from "@/api/lib/cache-memory"; @@ -46,7 +47,7 @@ import { localeRoutingFromConfig } from "@/lib/i18n/locale-routing"; import { buildApiMessagesSources } from "@/lib/i18n/sources"; import { resolvePersonalInformationFields } from "@/lib/user-personal-information"; import { manifest as coreManifest } from "@/manifest"; -import { realtime } from "@/ws/registry"; +import { realtime, runWithRealtimeKeepAlive } from "@/ws/registry"; import type { AiActionRegistry } from "../lib/ai/registry"; import type { BuildCronReturn } from "../lib/cron"; @@ -678,10 +679,20 @@ export const globalMiddleware = ({ c.set("adminSessionExpiresAt", null); c.set("log", loggerMiddleware(c)); - await next(); + // Realtime publishes started anywhere in this request - `c.get("realtime")` + // or a module that has no context of its own - are kept alive past the + // response, so a serverless instance does not freeze them half-sent. + await runWithRealtimeKeepAlive( + task => { + runAfterResponse(c, task); + }, + async () => { + await next(); - c.get("notifications").flushAfterResponse(); - c.get("email").flushAfterResponse(); + c.get("notifications").flushAfterResponse(); + c.get("email").flushAfterResponse(); + }, + ); }; }; diff --git a/packages/vitnode/src/api/middlewares/websocket-origin.middleware.test.ts b/packages/vitnode/src/api/middlewares/websocket-origin.middleware.test.ts index f5221bee1..3802d14be 100644 --- a/packages/vitnode/src/api/middlewares/websocket-origin.middleware.test.ts +++ b/packages/vitnode/src/api/middlewares/websocket-origin.middleware.test.ts @@ -6,12 +6,13 @@ import { websocketOriginMiddleware } from "./websocket-origin.middleware"; const upgrade = async ( origin: string | undefined, options?: { origin?: string[] }, + url = "/ws", ): Promise => { const app = new Hono(); app.get("/ws", websocketOriginMiddleware(options), c => c.text("upgraded")); return await app.request( - "/ws", + url, origin === undefined ? {} : { headers: { origin } }, ); }; @@ -73,6 +74,26 @@ describe("websocketOriginMiddleware", () => { expect(res.status).toBe(403); }); + it("allows a page on the handshake's own host, as a preview deployment is", async () => { + const res = await upgrade( + "https://preview-123.example", + undefined, + "https://preview-123.example/ws", + ); + + expect(res.status).toBe(200); + }); + + it("refuses another site even on the handshake's own path", async () => { + const res = await upgrade( + "https://evil.example", + undefined, + "https://preview-123.example/ws", + ); + + expect(res.status).toBe(403); + }); + it("allows a handshake with no Origin at all", async () => { // Browsers always send one, so this is a non-browser client - which carries // no ambient cookies to be ridden. diff --git a/packages/vitnode/src/api/middlewares/websocket-origin.middleware.ts b/packages/vitnode/src/api/middlewares/websocket-origin.middleware.ts index b96194ace..8ec86a1e8 100644 --- a/packages/vitnode/src/api/middlewares/websocket-origin.middleware.ts +++ b/packages/vitnode/src/api/middlewares/websocket-origin.middleware.ts @@ -35,7 +35,10 @@ const originOf = (value: string | undefined): string | undefined => { * WebSocket handshake is a `GET`. * * The `Origin` header is the defence, and it is a sound one here precisely - * because the browser sets it and a page cannot override it. A handshake that + * because the browser sets it and a page cannot override it. Allowed are the + * configured web and API origins, any extra `origin`, and the handshake's own + * origin - a page served by the same host, as on a single app's preview + * deployment, which a cross-site page cannot fake. A handshake that * carries **no** `Origin` is allowed through: browsers always send one, so an * originless handshake is a non-browser client, which has no cookies to be * ridden in the first place. @@ -62,7 +65,11 @@ export const websocketOriginMiddleware = ({ // No `Origin` at all: not a browser, so not a cross-site ride on somebody's // cookies. See the note above. - if (requestOrigin && !allowed.has(requestOrigin)) { + if ( + requestOrigin && + !allowed.has(requestOrigin) && + requestOrigin !== new URL(c.req.url).origin + ) { throw new HTTPException(403, { message: "WebSocket connections are not allowed from this origin.", }); diff --git a/packages/vitnode/src/framework/vite/dev-error.ts b/packages/vitnode/src/framework/vite/dev-error.ts index 08cd935f2..62a864793 100644 --- a/packages/vitnode/src/framework/vite/dev-error.ts +++ b/packages/vitnode/src/framework/vite/dev-error.ts @@ -32,7 +32,7 @@ const withNitroDevError = (): UserConfig => { }, }; - return config as UserConfig; + return config; }; /** diff --git a/packages/vitnode/src/framework/vite/env.test.ts b/packages/vitnode/src/framework/vite/env.test.ts index a5ea173c1..e1c3935b9 100644 --- a/packages/vitnode/src/framework/vite/env.test.ts +++ b/packages/vitnode/src/framework/vite/env.test.ts @@ -17,6 +17,7 @@ const ENV_FILE = [ const TOUCHED = [ "CRON_SECRET", "VITNODE_API_URL", + "VITNODE_EMBEDDED_WEBSOCKET", "VITNODE_EXTRA", "VITNODE_UNLISTED", "VITNODE_WEB_URL", @@ -89,10 +90,17 @@ describe("vitNodeEnv", () => { it("inlines the API and web URLs into the client bundle", async () => { expect(clientDefine(await runConfig(root))).toStrictEqual({ "process.env.VITNODE_API_URL": '"https://api.example.test"', + "process.env.VITNODE_EMBEDDED_WEBSOCKET": "undefined", "process.env.VITNODE_WEB_URL": '"https://web.example.test"', }); }); + it("tells the browser an app with webSocket serves its own socket", async () => { + const define = clientDefine(await runConfig(root, { webSocket: true })); + + expect(define["process.env.VITNODE_EMBEDDED_WEBSOCKET"]).toBe('"true"'); + }); + it("publishes nothing to the client beyond the listed keys", async () => { // The point of an explicit list. A secret reaching a browser bundle is not // recoverable by rotating a build, and `VITNODE_UNLISTED` shows that @@ -123,6 +131,7 @@ describe("vitNodeEnv", () => { // instead of falling through to the default the core config has for it. expect(clientDefine(await runConfig(root))).toStrictEqual({ "process.env.VITNODE_API_URL": "undefined", + "process.env.VITNODE_EMBEDDED_WEBSOCKET": "undefined", "process.env.VITNODE_WEB_URL": "undefined", }); }); @@ -138,6 +147,7 @@ describe("vitNodeEnv", () => { clientDefine(await runConfig(root, { clientEnv: ["VITNODE_EXTRA"] })), ).toStrictEqual({ "process.env.VITNODE_API_URL": '"https://api.example.test"', + "process.env.VITNODE_EMBEDDED_WEBSOCKET": "undefined", "process.env.VITNODE_EXTRA": '"from-the-app"', "process.env.VITNODE_WEB_URL": '"https://web.example.test"', }); @@ -162,6 +172,7 @@ describe("vitNodeEnv", () => { expect(Object.keys(define)).toStrictEqual([ "process.env.VITNODE_API_URL", "process.env.VITNODE_WEB_URL", + "process.env.VITNODE_EMBEDDED_WEBSOCKET", ]); }); @@ -173,6 +184,7 @@ describe("vitNodeEnv", () => { expect(Object.keys(define)).toStrictEqual([ "process.env.VITNODE_API_URL", "process.env.VITNODE_WEB_URL", + "process.env.VITNODE_EMBEDDED_WEBSOCKET", ]); }); }); diff --git a/packages/vitnode/src/framework/vite/env.ts b/packages/vitnode/src/framework/vite/env.ts index 7e4dd720c..a270c207c 100644 --- a/packages/vitnode/src/framework/vite/env.ts +++ b/packages/vitnode/src/framework/vite/env.ts @@ -11,6 +11,15 @@ export interface VitNodeEnvOptions { * values end up in the client bundle - never list a secret. */ clientEnv?: readonly string[]; + /** + * Whether this app serves `/api/ws` itself - an app that embeds the API. + * Mounts VitNode's Nitro WebSocket route and turns on the browser's + * WebSocket connection, which an app with an embedded API otherwise skips. + * The embedded API still needs its `/ws` handshake route. An app talking to a + * separate API (`VITNODE_API_URL`) connects to it regardless. Defaults to + * `false`. + */ + webSocket?: boolean; } /** @@ -19,6 +28,7 @@ export interface VitNodeEnvOptions { */ export const vitNodeEnv = ({ clientEnv = [], + webSocket = false, }: VitNodeEnvOptions = {}): Plugin => { // De-duplicated, so an app that names one of the package's own keys gets one // `define` entry rather than a silently repeated one. @@ -36,12 +46,13 @@ export const vitNodeEnv = ({ for (const [key, value] of Object.entries(env)) { process.env[key] ??= value; } + if (webSocket) process.env.VITNODE_EMBEDDED_WEBSOCKET ??= "true"; return { environments: { client: { define: Object.fromEntries( - keys.map(key => [ + [...keys, "VITNODE_EMBEDDED_WEBSOCKET"].map(key => [ `process.env.${key}`, // `undefined` when unset rather than nothing at all: the read // has to be replaced either way, or it throws in the browser diff --git a/packages/vitnode/src/framework/vite/index.ts b/packages/vitnode/src/framework/vite/index.ts index bf28fd7ed..6493ce4dd 100644 --- a/packages/vitnode/src/framework/vite/index.ts +++ b/packages/vitnode/src/framework/vite/index.ts @@ -12,3 +12,4 @@ export type { VitNodeTailwindSourcesOptions } from "./tailwind-sources"; export { vitNodeTailwindSources } from "./tailwind-sources"; export type { VitNodeViteOptions } from "./vitnode"; export { vitnode } from "./vitnode"; +export { vitNodeWebSocket } from "./websocket"; diff --git a/packages/vitnode/src/framework/vite/prerender-routes.ts b/packages/vitnode/src/framework/vite/prerender-routes.ts index add0e2de0..7702e4c8e 100644 --- a/packages/vitnode/src/framework/vite/prerender-routes.ts +++ b/packages/vitnode/src/framework/vite/prerender-routes.ts @@ -25,15 +25,19 @@ export class PrerenderDomainsError extends Error { /** * Expands canonical paths into the public URL of each enabled locale - * prefixes and translated `routePaths` included - for Nitro's - * `prerender.routes`. Duplicates are removed. + * `prerender.routes`, passed through the `nitro` option of `vitnode()`. + * Duplicates are removed. * * @throws {@link PrerenderDomainsError} when `i18n.domains` is set. * * @example * ```ts - * nitro({ - * prerender: { - * routes: prerenderRoutes({ i18n: vitNodeConfig.i18n, paths: ["/", "/about"] }), + * vitnode({ + * appRoot: import.meta.dirname, + * nitro: { + * prerender: { + * routes: prerenderRoutes({ i18n: vitNodeConfig.i18n, paths: ["/", "/about"] }), + * }, * }, * }); * ``` diff --git a/packages/vitnode/src/framework/vite/vitnode.test.ts b/packages/vitnode/src/framework/vite/vitnode.test.ts index ed975ad73..f50fe4cad 100644 --- a/packages/vitnode/src/framework/vite/vitnode.test.ts +++ b/packages/vitnode/src/framework/vite/vitnode.test.ts @@ -5,15 +5,19 @@ import { vitnode } from "./vitnode"; const names = (plugins: { name: string }[]) => plugins.map(({ name }) => name); describe("the composed plugin", () => { - it("returns all six, in the order they have to run", () => { - expect(names(vitnode({ appRoot: "/app" }))).toEqual([ + it("returns VitNode's seven in the order they have to run, then Nitro", () => { + const plugins = names(vitnode({ appRoot: "/app" })); + + expect(plugins.slice(0, 7)).toEqual([ "vitnode:env", "vitnode:optimize-deps", "vitnode:ssr-externals", "vitnode:tailwind-sources", "vitnode:plugin-routes", "vitnode:dev-error", + "vitnode:websocket", ]); + expect(plugins.slice(7)).toContain("nitro:main"); }); it("needs nothing but the app root", () => { diff --git a/packages/vitnode/src/framework/vite/vitnode.ts b/packages/vitnode/src/framework/vite/vitnode.ts index 8b3f61626..281c7952f 100644 --- a/packages/vitnode/src/framework/vite/vitnode.ts +++ b/packages/vitnode/src/framework/vite/vitnode.ts @@ -1,5 +1,8 @@ +import type { NitroPluginConfig } from "nitro/vite"; import type { Plugin } from "vite"; +import { nitro as nitroPlugin } from "nitro/vite"; + import type { VitNodeEnvOptions } from "./env"; import type { VitNodePluginRoutesOptions } from "./plugin-routes"; @@ -9,16 +12,27 @@ import { vitNodeOptimizeDeps } from "./optimize-deps"; import { vitNodePluginRoutes } from "./plugin-routes"; import { vitNodeSsrExternals } from "./ssr-externals"; import { vitNodeTailwindSources } from "./tailwind-sources"; +import { vitNodeWebSocket } from "./websocket"; /** Options of the {@link vitnode} Vite plugin. */ export interface VitNodeViteOptions - extends VitNodeEnvOptions, VitNodePluginRoutesOptions {} + extends VitNodeEnvOptions, VitNodePluginRoutesOptions { + /** + * Options for Nitro, the server the app is built for and served by - + * `prerender`, `preset`, `rollupConfig`, `routeRules`, `handlers` and the + * rest of Nitro's config. VitNode adds its own on top: the SSR externals in a + * build, the development error page in `vite dev` and, with `webSocket`, the + * `/api/ws` route. Arrays are merged, so your `handlers` are kept alongside + * VitNode's. + */ + nitro?: NitroPluginConfig; +} /** * VitNode's Vite plugins for a TanStack Start app, in one call: environment * variables, dependency pre-bundling, SSR externals, Tailwind sources for the - * installed packages, the generated plugin registries, and the development - * error page. + * installed packages, the generated plugin registries, the development error + * page, the WebSocket route and Nitro itself - do not add `nitro()` again. * * Add it before `tanstackStart()` in `vite.config.ts`. * @@ -29,7 +43,13 @@ export interface VitNodeViteOptions * import { defineConfig } from "vite"; * * export default defineConfig({ - * plugins: [vitnode({ appRoot: import.meta.dirname }), tanstackStart()], + * plugins: [ + * vitnode({ + * appRoot: import.meta.dirname, + * nitro: { rollupConfig: { external: [/^@sentry\//] } }, + * }), + * tanstackStart(), + * ], * }); * ``` */ @@ -37,11 +57,15 @@ export const vitnode = ({ appRoot, clientEnv, hostRoutesDir, + nitro, + webSocket, }: VitNodeViteOptions): Plugin[] => [ - vitNodeEnv({ clientEnv }), + vitNodeEnv({ clientEnv, webSocket }), vitNodeOptimizeDeps(), vitNodeSsrExternals({ appRoot }), vitNodeTailwindSources({ appRoot }), vitNodePluginRoutes({ appRoot, hostRoutesDir }), vitNodeDevError(), + vitNodeWebSocket({ webSocket }), + ...nitroPlugin(nitro), ]; diff --git a/packages/vitnode/src/framework/vite/websocket.ts b/packages/vitnode/src/framework/vite/websocket.ts new file mode 100644 index 000000000..d0fe954ae --- /dev/null +++ b/packages/vitnode/src/framework/vite/websocket.ts @@ -0,0 +1,43 @@ +import type { Plugin, UserConfig } from "vite"; + +import { fileURLToPath } from "node:url"; + +const WEBSOCKET_ROUTE = "/api/ws"; +const WEBSOCKET_ROUTE_MODULE = "@vitnode/core/ws/nitro-route"; + +interface NitroWebSocketConfig { + nitro: { + features: { websocket: true }; + handlers: { handler: string; route: string }[]; + noExternals: string[]; + }; +} + +const withNitroWebSocket = (): UserConfig => { + const handler = fileURLToPath(import.meta.resolve(WEBSOCKET_ROUTE_MODULE)); + const config: NitroWebSocketConfig = { + nitro: { + features: { websocket: true }, + handlers: [{ handler, route: WEBSOCKET_ROUTE }], + noExternals: [handler], + }, + }; + + return config; +}; + +/** + * Mounts VitNode's WebSocket at `/api/ws` as a Nitro route and turns on + * Nitro's WebSocket support. The route hands each handshake to the API + * embedded in the app's server bundle, and is always bundled into Nitro's + * output, so it reaches Nitro's runtime in a prerender and in `vite dev` too. + * Does nothing when `webSocket` is off. + */ +export const vitNodeWebSocket = ({ + webSocket = false, +}: { + webSocket?: boolean; +} = {}): Plugin => ({ + config: () => (webSocket ? withNitroWebSocket() : undefined), + name: "vitnode:websocket", +}); diff --git a/packages/vitnode/src/lib/docs-links.ts b/packages/vitnode/src/lib/docs-links.ts index d54ad6e00..dd36d76ad 100644 --- a/packages/vitnode/src/lib/docs-links.ts +++ b/packages/vitnode/src/lib/docs-links.ts @@ -8,7 +8,7 @@ export const DOCS_URLS = { queue: "https://vitnode.com/docs/dev/advanced/queue", redis: "https://vitnode.com/docs/dev/advanced/redis", storage: "https://vitnode.com/docs/dev/storage", - websocket: "https://vitnode.com/docs/dev/websocket", + websocket: "https://vitnode.com/docs/dev/websocket/setup", }; export const VITNODE_WEBSITE_URL = "https://vitnode.com/"; diff --git a/packages/vitnode/src/views/admin/views/content/live/use-session.test.tsx b/packages/vitnode/src/views/admin/views/content/live/use-session.test.tsx index fe9ad2053..914844586 100644 --- a/packages/vitnode/src/views/admin/views/content/live/use-session.test.tsx +++ b/packages/vitnode/src/views/admin/views/content/live/use-session.test.tsx @@ -407,6 +407,81 @@ describe("useContentLiveSession", () => { }); }); + describe("in a tab that follows another tab's socket", () => { + class FakeBroadcastChannel { + constructor() { + channels.push(this); + } + + onmessage: ((event: { data: unknown }) => void) | null = null; + posted: unknown[] = []; + + close() { + this.onmessage = null; + } + + postMessage(message: unknown) { + this.posted.push(message); + } + } + + let channels: FakeBroadcastChannel[] = []; + + const leaderOpened = (connectionId: string) => { + act(() => { + channels[0]?.onmessage?.({ + data: { connectionId, payload: 1, type: "ready-state" }, + }); + }); + }; + + const joinsRelayed = () => + (channels[0]?.posted ?? []).flatMap(message => { + const { payload, type } = message as { + payload?: { data: ContentLiveClientMessage }; + type: string; + }; + + return type === "send" && payload?.data.type === "join" + ? [payload.data] + : []; + }); + + beforeEach(() => { + channels = []; + sockets = []; + vi.stubEnv("VITNODE_API_URL", "http://localhost:8000"); + vi.stubGlobal("WebSocket", FakeWebSocket); + vi.stubGlobal("BroadcastChannel", FakeBroadcastChannel); + // The leader tab holds the lock, so this one never opens a socket. + vi.stubGlobal("navigator", { + locks: { request: async () => await new Promise(() => {}) }, + onLine: true, + userAgent: navigator.userAgent, + }); + }); + + afterEach(() => { + vi.unstubAllEnvs(); + vi.unstubAllGlobals(); + }); + + it("joins again through the leader for every new connection, though it never saw the socket close", async () => { + await mount(); + + leaderOpened("connection-1"); + leaderOpened("connection-2"); + + const joins = joinsRelayed(); + expect(sockets).toHaveLength(0); + expect(joins).toEqual([ + expect.objectContaining({ room: ROOM, type: "join" }), + expect.objectContaining({ room: ROOM, type: "join" }), + ]); + expect(joins[1]?.clientId).toBe(joins[0]?.clientId); + }); + }); + describe("without a socket", () => { beforeEach(() => { vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] }); diff --git a/packages/vitnode/src/views/admin/views/content/live/use-session.ts b/packages/vitnode/src/views/admin/views/content/live/use-session.ts index 9adb61bb2..2a2c92663 100644 --- a/packages/vitnode/src/views/admin/views/content/live/use-session.ts +++ b/packages/vitnode/src/views/admin/views/content/live/use-session.ts @@ -220,16 +220,19 @@ export const useContentLiveSession = ({ } }; - const { readyState, send } = useVitNodeWebSocket(contentLiveChannel, { - onMessage, - }); + const { connectionId, readyState, send } = useVitNodeWebSocket( + contentLiveChannel, + { onMessage }, + ); const readyStateRef = React.useRef(readyState); React.useEffect(() => { readyStateRef.current = readyState; }, [readyState]); + // Every new connection starts with no rooms on the server - a reconnect, a + // new leader tab or a serverless duration limit - so each one joins again. React.useEffect(() => { - if (readyState !== 1) return; + if (connectionId === null) return; joiningRef.current = true; send({ clientId, locale: languageRef.current, room, type: "join" }); @@ -238,7 +241,7 @@ export const useContentLiveSession = ({ joiningRef.current = false; setJoined(false); }; - }, [clientId, readyState, room, send]); + }, [clientId, connectionId, room, send]); React.useEffect(() => { const leave = () => { diff --git a/packages/vitnode/src/ws/crossws.server.test.ts b/packages/vitnode/src/ws/crossws.server.test.ts new file mode 100644 index 000000000..b39e4d253 --- /dev/null +++ b/packages/vitnode/src/ws/crossws.server.test.ts @@ -0,0 +1,382 @@ +import type { WSContext, WSEvents } from "hono/ws"; + +import { Hono } from "hono"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import type { WebSocketConfig } from "@/api/lib/websocket"; +import type { + EnvVariablesVitNode, + EnvVitNode, +} from "@/api/middlewares/global.middleware"; + +import { buildWebSocket } from "@/api/lib/websocket"; +import { websocketOriginMiddleware } from "@/api/middlewares/websocket-origin.middleware"; +import { core_notification_user_state } from "@/database/notifications"; +import { configureInProcessApi } from "@/lib/fetcher/in-process-api.server"; +import { createMemoryDb } from "@/tests/memory-db"; + +import type { VitNodeCrosswsHooks, VitNodeCrosswsPeer } from "./crossws.server"; + +import { + createVitNodeWebSocketHooks, + upgradeWebSocket, +} from "./crossws.server"; +import { handleVitNodeWebSocket } from "./handle"; +import { notificationsStateChannel } from "./notifications"; +import { wsRegistry } from "./registry"; +import { getWebSocketId, WS_PING_ID, WS_PONG_ID } from "./types"; + +const ORIGIN = "https://site.example"; +const WS_URL = `${ORIGIN}/api/ws`; + +const handled: { data: unknown; ws: WSContext }[] = []; + +const roomsSocket: WebSocketConfig = { + ...buildWebSocket<{ room: string }, { joined: string }>({ + id: "rooms", + onMessage: ({ data, send, ws }) => { + handled.push({ data, ws }); + wsRegistry.join(ws, data.room, "tab-a"); + send({ joined: data.room }); + }, + }), + module: "test", + pluginId: "@vitnode/test", +}; +const ROOMS_ID = getWebSocketId(roomsSocket); + +/** + * The embedded API's `/api/ws` route as an app mounts it, behind a stand-in for + * the global middleware: the `user` cookie plays the session. + */ +const buildApp = () => { + const inline = { + onClose: vi.fn>(), + onError: vi.fn>(), + onMessage: vi.fn>(), + }; + const createEvents = vi.fn(() => inline); + const { db } = createMemoryDb([ + [ + core_notification_user_state, + [{ revision: 3, unreadCount: 2, userId: 7 }], + ], + ]); + + const app = new Hono().basePath("/api"); + app.use("*", async (c, next) => { + c.set("core", { + webSockets: [roomsSocket], + } as unknown as EnvVariablesVitNode["core"]); + c.set("db", db as unknown as EnvVariablesVitNode["db"]); + const userId = /(?:^|; )user=(\d+)/.exec(c.req.header("cookie") ?? "")?.[1]; + c.set( + "user", + userId + ? ({ id: Number(userId) } as unknown as EnvVariablesVitNode["user"]) + : null, + ); + if (c.req.header("x-refresh-session")) { + c.header("set-cookie", "session=fresh; Path=/", { append: true }); + c.header("set-cookie", "csrf=token; Path=/", { append: true }); + } + await next(); + }); + app.get( + "/ws", + websocketOriginMiddleware(), + upgradeWebSocket(handleVitNodeWebSocket(createEvents)), + ); + + return { app, createEvents, inline }; +}; + +const handshake = (headers: Record = {}) => + new Request(WS_URL, { headers: { origin: ORIGIN, ...headers } }); + +const peers: { hooks: VitNodeCrosswsHooks; peer: VitNodeCrosswsPeer }[] = []; + +/** The accepted upgrade, or a failure naming the status it was refused with. */ +const accept = async (upgrade: ReturnType) => { + const result = await upgrade; + if (result instanceof Response) { + throw new Error(`Upgrade refused with ${result.status}`); + } + + return result; +}; + +const fakePeer = (context: Record) => { + const sent: unknown[] = []; + const peer = { + close: vi.fn(), + // crossws merges the context `upgrade` returned into the peer's own. + context: { ...context }, + request: { url: WS_URL }, + send: (data: ArrayBuffer | string | Uint8Array) => { + sent.push(typeof data === "string" ? JSON.parse(data) : data); + }, + websocket: { readyState: 1 }, + } satisfies VitNodeCrosswsPeer; + + return { peer, sent }; +}; + +const textMessage = (rawData: unknown) => ({ + arrayBuffer: () => new ArrayBuffer(0), + rawData, +}); + +const setup = () => { + const built = buildApp(); + const hooks = createVitNodeWebSocketHooks({ + api: async request => await built.app.fetch(request), + }); + + const connect = async (headers: Record = {}) => { + const { context } = await accept(hooks.upgrade(handshake(headers))); + const client = fakePeer(context); + peers.push({ hooks, peer: client.peer }); + hooks.open(client.peer); + + return { + ...client, + say: (message: unknown) => { + hooks.message(client.peer, textMessage(JSON.stringify(message))); + }, + }; + }; + + return { ...built, connect, hooks }; +}; + +beforeEach(() => { + handled.length = 0; +}); + +afterEach(() => { + peers.splice(0).forEach(({ hooks, peer }) => { + hooks.close(peer, { code: 1000 }); + }); + vi.restoreAllMocks(); +}); + +describe("the shared protocol through crossws", () => { + it("answers the client's ping with a pong", async () => { + const { connect } = setup(); + const client = await connect(); + + client.say({ data: null, id: WS_PING_ID }); + + expect(client.sent).toEqual([{ data: null, id: WS_PONG_ID }]); + }); + + it("routes messages to their socket's handler with one stable connection object", async () => { + const { connect } = setup(); + const client = await connect(); + + client.say({ data: { room: "a" }, id: ROOMS_ID }); + client.say({ data: { room: "b" }, id: ROOMS_ID }); + + expect(handled.map(({ data }) => data)).toEqual([ + { room: "a" }, + { room: "b" }, + ]); + expect(handled[0]?.ws).toBe(handled[1]?.ws); + expect(wsRegistry.roomsOf(handled[0].ws)).toEqual(["a", "b"]); + expect(client.sent).toEqual([ + { data: { joined: "a" }, id: ROOMS_ID }, + { data: { joined: "b" }, id: ROOMS_ID }, + ]); + }); + + it("hands unmatched, non-JSON and binary messages to the connection's own handler", async () => { + const { connect, hooks, inline } = setup(); + const client = await connect(); + const bytes = new Uint8Array([1, 2, 3]).buffer; + + client.say({ data: 1, id: "@vitnode/test_test_missing" }); + hooks.message(client.peer, textMessage("not json")); + hooks.message(client.peer, { + arrayBuffer: () => bytes, + rawData: new Uint8Array(bytes), + }); + + expect(inline.onMessage.mock.calls.map(([event]) => event.data)).toEqual([ + JSON.stringify({ data: 1, id: "@vitnode/test_test_missing" }), + "not json", + bytes, + ]); + expect(handled).toEqual([]); + expect(client.sent).toEqual([]); + }); +}); + +describe("the handshake", () => { + it("rejects another site with the API's 403 before any connection events exist", async () => { + const { createEvents, hooks } = setup(); + + const rejection: unknown = await hooks + .upgrade(handshake({ origin: "https://evil.example" })) + .catch((error: unknown) => error); + + expect(rejection).toBeInstanceOf(Response); + expect((rejection as Response).status).toBe(403); + expect(createEvents).not.toHaveBeenCalled(); + }); + + it("closes a peer that opens without an accepted handshake and never registers it", () => { + const { hooks } = setup(); + const { peer, sent } = fakePeer({}); + + hooks.open(peer); + hooks.message(peer, textMessage(JSON.stringify({ id: WS_PING_ID }))); + wsRegistry.broadcast("test", 1); + + expect(peer.close).toHaveBeenCalledWith(1011, "Missing handshake"); + expect(sent).toEqual([]); + }); + + it("answers a plain GET of the route with 426", async () => { + const { app } = setup(); + + const res = await app.request(WS_URL, { headers: { origin: ORIGIN } }); + + expect(res.status).toBe(426); + }); + + it("passes the session cookies the handshake set on to the upgrade", async () => { + const { hooks } = setup(); + + const { headers } = await accept( + hooks.upgrade(handshake({ "x-refresh-session": "1" })), + ); + + expect(headers).toEqual([ + ["set-cookie", expect.stringContaining("session=fresh")], + ["set-cookie", expect.stringContaining("csrf=token")], + ]); + }); + + it("registers a signed-in connection as its user and a guest as nobody", async () => { + const { connect } = setup(); + const member = await connect({ cookie: "user=7" }); + const guest = await connect(); + + wsRegistry.sendToUser(7, "inbox", { n: 1 }); + + expect(member.sent).toContainEqual({ data: { n: 1 }, id: "inbox" }); + expect(guest.sent).toEqual([]); + }); + + it("sends a signed-in connection its notification state as it opens", async () => { + const { connect } = setup(); + + const member = await connect({ cookie: "user=7" }); + + await vi.waitFor(() => { + expect(member.sent).toEqual([ + { + data: { reason: "reconciled", revision: 3, unread: 2 }, + id: notificationsStateChannel.id, + }, + ]); + }); + }); +}); + +describe("on a cold instance with no API registered yet", () => { + afterEach(() => { + configureInProcessApi(undefined as never); + }); + + it("answers 503 when loading the app registers no API", async () => { + const warmup = vi.fn(async () => await Promise.resolve()); + vi.spyOn(console, "error").mockImplementation(() => undefined); + const hooks = createVitNodeWebSocketHooks({ warmup }); + + const rejection: unknown = await hooks + .upgrade(handshake()) + .catch((error: unknown) => error); + + expect(warmup).toHaveBeenCalledTimes(1); + expect(rejection).toBeInstanceOf(Response); + expect((rejection as Response).status).toBe(503); + }); + + it("loads the app once for concurrent handshakes, then authenticates through it", async () => { + const { app } = buildApp(); + const warmup = vi.fn(async (_request: Request) => { + await Promise.resolve(); + configureInProcessApi(async request => await app.fetch(request)); + }); + const hooks = createVitNodeWebSocketHooks({ warmup }); + + const accepted = await Promise.all([ + accept(hooks.upgrade(handshake())), + accept(hooks.upgrade(handshake())), + ]); + + expect(warmup).toHaveBeenCalledTimes(1); + expect(warmup.mock.calls[0]?.[0]).toMatchObject({ + method: "GET", + url: WS_URL, + }); + expect(accepted).toHaveLength(2); + }); +}); + +describe("closing a connection", () => { + it.each([ + [ + "closes", + (hooks: VitNodeCrosswsHooks, peer: VitNodeCrosswsPeer) => { + hooks.close(peer, { code: 1000 }); + }, + ], + [ + "errors and then closes", + (hooks: VitNodeCrosswsHooks, peer: VitNodeCrosswsPeer) => { + hooks.error(peer, new Error("reset")); + hooks.close(peer, { code: 1006 }); + }, + ], + ])( + "leaves the registry and every room exactly once when the socket %s", + async (_, end) => { + vi.spyOn(console, "error").mockImplementation(() => undefined); + const { connect, hooks, inline } = setup(); + const client = await connect({ cookie: "user=8" }); + await vi.waitFor(() => { + expect(client.sent).toHaveLength(1); + }); + client.say({ data: { room: "a" }, id: ROOMS_ID }); + const ws = handled[0].ws; + const left = vi.fn(); + const stop = wsRegistry.onConnectionClose(left); + + end(hooks, client.peer); + stop(); + const sentBefore = client.sent.length; + wsRegistry.sendToUser(8, "inbox", 1); + client.say({ data: null, id: WS_PING_ID }); + + expect(left).toHaveBeenCalledTimes(1); + expect(left).toHaveBeenCalledWith(ws, [{ memberId: "tab-a", room: "a" }]); + expect(inline.onClose).toHaveBeenCalledTimes(1); + expect(wsRegistry.membersOf("a")).toEqual([]); + expect(client.sent).toHaveLength(sentBefore); + }, + ); + + it("closes the peer with 1011 and tells the connection's onError on a runtime error", async () => { + vi.spyOn(console, "error").mockImplementation(() => undefined); + const { connect, hooks, inline } = setup(); + const client = await connect(); + + hooks.error(client.peer, new Error("reset")); + + expect(inline.onError).toHaveBeenCalledTimes(1); + expect(client.peer.close).toHaveBeenCalledWith(1011, "Internal error"); + }); +}); diff --git a/packages/vitnode/src/ws/crossws.server.ts b/packages/vitnode/src/ws/crossws.server.ts new file mode 100644 index 000000000..eaf9dfc0a --- /dev/null +++ b/packages/vitnode/src/ws/crossws.server.ts @@ -0,0 +1,338 @@ +import type { WSEvents, WSReadyState } from "hono/ws"; + +import { defineWebSocketHelper, WSContext } from "hono/ws"; + +import type { InProcessApiHandler } from "@/lib/fetcher/in-process-api.server"; + +import { getInProcessApi } from "@/lib/fetcher/in-process-api.server"; + +// `Symbol.for`, not a module-level symbol: the Hono route below runs in the +// app's server bundle and the hooks in Nitro's, and each bundle has its own +// copy of this module. Both read the same key off the same `Request`. +const HANDSHAKE_KEY = Symbol.for("vitnode.ws.crossws.handshake"); + +const CONTEXT_KEY = "vitnodeWebSocket"; + +interface Handshake { + events?: WSEvents; +} + +type HandshakeRequest = Request & { [HANDSHAKE_KEY]?: Handshake }; + +/** + * The Hono helper for a `/ws` route served through Nitro (crossws). It does not + * upgrade anything itself: it hands the connection's events, created after the + * API's middleware has run, back to {@link createVitNodeWebSocketHooks}, which + * performs the upgrade. The counterpart of `upgradeWebSocket` from + * `@hono/node-server` in a standalone API. Server-only. + * + * A request that did not come from those hooks - a plain `GET /api/ws` - gets + * `426 Upgrade Required`. + * + * @example + * ```ts + * import { websocketOriginMiddleware } from "@vitnode/core/api/middlewares/websocket-origin.middleware"; + * import { upgradeWebSocket } from "@vitnode/core/ws/crossws.server"; + * import { handleVitNodeWebSocket } from "@vitnode/core/ws/handle"; + * + * app.get( + * "/ws", + * websocketOriginMiddleware(), + * upgradeWebSocket(handleVitNodeWebSocket()), + * ); + * ``` + */ +export const upgradeWebSocket = defineWebSocketHelper((c, events) => { + const handshake = (c.req.raw as HandshakeRequest)[HANDSHAKE_KEY]; + if (!handshake) return c.text("WebSocket upgrade is required.", 426); + + handshake.events = events; + + return c.body(null, 200); +}); + +/** The parts of a crossws peer the adapter uses. */ +export interface VitNodeCrosswsPeer { + close: (code?: number, reason?: string) => void; + readonly context: Record; + readonly request: { readonly url: string }; + send: (data: ArrayBuffer | string | Uint8Array) => unknown; + readonly websocket: { + readonly protocol?: string; + readonly readyState?: number; + }; +} + +/** The parts of a crossws message the adapter uses. */ +export interface VitNodeCrosswsMessage { + arrayBuffer: () => ArrayBuffer | SharedArrayBuffer; + readonly rawData: unknown; +} + +/** The crossws hooks {@link createVitNodeWebSocketHooks} returns. */ +export interface VitNodeCrosswsHooks { + close: ( + peer: VitNodeCrosswsPeer, + details: { code?: number; reason?: string }, + ) => void; + error: (peer: VitNodeCrosswsPeer, error: unknown) => void; + message: (peer: VitNodeCrosswsPeer, message: VitNodeCrosswsMessage) => void; + open: (peer: VitNodeCrosswsPeer) => void; + /** Resolves the rejection as a `Response`, or the accepted connection. */ + upgrade: (request: Request) => Promise< + | Response + | { + context: Record; + headers: [string, string][]; + } + >; +} + +export interface VitNodeWebSocketHooksOptions { + /** + * The in-process API to authenticate handshakes with. Defaults to the one + * the app registered with `configureInProcessApi`. + */ + api?: InProcessApiHandler; + /** + * Loads the app's server bundle when no API has registered yet - a cold + * instance whose first request is a WebSocket upgrade. Receives a plain + * `GET` for the same URL; its response is ignored. In a Nitro app pass + * `request => fetchViteEnv("ssr", request)` from `nitro/vite/runtime`. + */ + warmup?: (request: Request) => Promise; +} + +interface Connection { + closed: boolean; + events: WSEvents; + ws?: WSContext; +} + +const logError = (message: string, error: unknown): void => { + // oxlint-disable-next-line no-console + console.error(message, error); +}; + +const connectionOf = (peer: VitNodeCrosswsPeer): Connection | undefined => { + const connection = peer.context[CONTEXT_KEY]; + + return typeof connection === "object" && connection !== null + ? (connection as Connection) + : undefined; +}; + +const clientAddressOf = (request: Request): string | undefined => { + const ip: unknown = Reflect.get(request, "ip"); + + return typeof ip === "string" && ip.length > 0 ? ip : undefined; +}; + +const closeEvent = (code?: number, reason?: string): CloseEvent => { + const init = { code, reason, wasClean: code !== undefined && code !== 1006 }; + if (typeof CloseEvent === "function") return new CloseEvent("close", init); + + // Node 22 has no `CloseEvent` global. + return Object.assign(new Event("close"), init) as CloseEvent; +}; + +/** + * Runs the connection's `onClose` exactly once, whether the socket closed, the + * runtime reported an error, or both - so it leaves the registry and every room + * once. + */ +const finish = ( + connection: Connection, + details: { code?: number; reason?: string }, +): void => { + if (connection.closed) return; + connection.closed = true; + if (!connection.ws) return; + + try { + connection.events.onClose?.( + closeEvent(details.code, details.reason), + connection.ws, + ); + } catch (error) { + logError("WebSocket close handler error:", error); + } +}; + +/** + * crossws hooks that serve VitNode's `/api/ws` from Nitro, for an app that + * embeds the API (`configureInProcessApi`). Pass them to + * `defineWebSocketHandler` in a Nitro route. Server-only. + * + * Each handshake runs through the embedded Hono app as a `GET /api/ws`, so the + * Origin check, the rate limiter, the session cookie and every + * `c.get(...)` model are exactly what a standalone API's handshake gets; the + * route must use {@link upgradeWebSocket}. Anything but an accepted handshake + * is returned as the rejection `Response` and nothing is registered. Once open, the + * connection keeps one `WSContext` for its lifetime and runs the same + * `handleVitNodeWebSocket` events as `@hono/node-server`. + * + * Liveness is the runtime's job: crossws pings and drops silent sockets, and + * the client sends its own `vitnode_ping`. + * + * `vitnode({ webSocket: true })` from `@vitnode/core/framework/vite` already + * mounts a Nitro route at `/api/ws` that does this; write your own only for a + * different route. Import this module dynamically there, after the warmup: a + * static import bundles all of `@vitnode/core` into Nitro's entry, where it + * starts before the app has configured it. + * + * @example + * ```ts + * // a custom Nitro route, registered through `vitnode({ nitro: { handlers } })` + * import { defineWebSocketHandler } from "nitro"; + * import { fetchViteEnv } from "nitro/vite/runtime"; + * + * const warmup = async (request: Request) => fetchViteEnv("ssr", request); + * + * export default defineWebSocketHandler(async event => { + * await warmup(new Request(event.req.url)); + * const { createVitNodeWebSocketHooks } = + * await import("@vitnode/core/ws/crossws.server"); + * + * return createVitNodeWebSocketHooks({ warmup }); + * }); + * ``` + */ +export const createVitNodeWebSocketHooks = ({ + api, + warmup, +}: VitNodeWebSocketHooksOptions = {}): VitNodeCrosswsHooks => { + let warming: Promise | undefined; + + const resolveApi = async ( + request: Request, + ): Promise => { + const current = api ?? getInProcessApi(); + if (current || !warmup) return current; + + warming ??= warmup(new Request(request.url)) + .catch((error: unknown) => { + logError("WebSocket warmup error:", error); + }) + .finally(() => { + warming = undefined; + }); + await warming; + + return getInProcessApi(); + }; + + return { + close: (peer, details) => { + const connection = connectionOf(peer); + if (connection) finish(connection, details); + }, + error: (peer, error) => { + logError("WebSocket error:", error); + const connection = connectionOf(peer); + if (!connection) return; + + if (connection.ws) { + try { + connection.events.onError?.(new Event("error"), connection.ws); + } catch (handlerError) { + logError("WebSocket error handler error:", handlerError); + } + } + finish(connection, { code: 1011, reason: "Internal error" }); + try { + peer.close(1011, "Internal error"); + } catch { + // Already gone. + } + }, + message: (peer, message) => { + const connection = connectionOf(peer); + if (!connection?.ws || connection.closed) return; + + const data = + typeof message.rawData === "string" + ? message.rawData + : message.arrayBuffer(); + try { + connection.events.onMessage?.( + new MessageEvent("message", { data }), + connection.ws, + ); + } catch (error) { + logError("WebSocket message handler error:", error); + } + }, + open: peer => { + const connection = connectionOf(peer); + if (!connection) { + peer.close(1011, "Missing handshake"); + + return; + } + + // One wrapper for the connection's whole life: the registry, the rooms + // and every handler key their state by this object. + connection.ws = new WSContext({ + close: (code, reason) => { + peer.close(code, reason); + }, + protocol: peer.websocket.protocol, + raw: peer, + get readyState() { + return (peer.websocket.readyState ?? 1) as WSReadyState; + }, + send: data => { + peer.send(data); + }, + url: peer.request.url, + }); + + try { + connection.events.onOpen?.(new Event("open"), connection.ws); + } catch (error) { + logError("WebSocket open handler error:", error); + } + }, + upgrade: async request => { + const handler = await resolveApi(request); + if (!handler) { + return new Response("The VitNode API is not available.", { + status: 503, + }); + } + + const handshake: Handshake = {}; + const handshakeRequest: HandshakeRequest = new Request(request.url, { + headers: request.headers, + }); + handshakeRequest[HANDSHAKE_KEY] = handshake; + + const response = await handler(handshakeRequest, { + clientAddress: clientAddressOf(request), + }); + + if (!handshake.events || !response.ok) { + // Origin refused, rate limited, not ready, ... - the API's own answer. + // crossws sends a returned `Response` back instead of upgrading. + return response.ok + ? new Response("WebSocket upgrade failed.", { status: 500 }) + : response; + } + await response.body?.cancel(); + + return { + context: { + [CONTEXT_KEY]: { + closed: false, + events: handshake.events, + } satisfies Connection, + }, + // A handshake that refreshed the session passes the cookie on. + headers: response.headers + .getSetCookie() + .map((cookie): [string, string] => ["set-cookie", cookie]), + }; + }, + }; +}; diff --git a/packages/vitnode/src/ws/handle.ts b/packages/vitnode/src/ws/handle.ts index 66083073d..4024ebbb2 100644 --- a/packages/vitnode/src/ws/handle.ts +++ b/packages/vitnode/src/ws/handle.ts @@ -1,13 +1,19 @@ import type { Context } from "hono"; import type { WSContext, WSEvents } from "hono/ws"; +import { HTTPException } from "hono/http-exception"; + import type { EnvVitNode } from "@/api/middlewares/global.middleware"; import type { NotificationStateMessage } from "@/lib/notifications/types"; import type { VitNodeWSMessage } from "@/ws/types"; import { getNotificationState } from "@/api/models/notifications/inbox"; import { notificationsStateChannel } from "@/ws/notifications"; -import { markWebSocketEnabled, wsRegistry } from "@/ws/registry"; +import { + markWebSocketEnabled, + waitForRealtimePubSub, + wsRegistry, +} from "@/ws/registry"; import { getWebSocketId, WS_PING_ID, WS_PONG_ID } from "@/ws/types"; const parseMessage = (raw: unknown): undefined | VitNodeWSMessage => { @@ -37,6 +43,8 @@ const sendNotificationState = async ( userId: number, ): Promise => { const state = await getNotificationState(c.get("db"), userId); + // The socket may have closed while the state was read. + if (ws.readyState !== 1) return; ws.send( JSON.stringify({ data: { ...state, reason: "reconciled" }, @@ -48,12 +56,33 @@ const sendNotificationState = async ( /** * Builds the Hono `upgradeWebSocket` handler for the app's `/ws` route: tracks * each connection with its signed-in user and routes incoming messages to the - * matching `buildWebSocket` handler by id. + * matching `buildWebSocket` handler by id. The protocol is the same whichever + * runtime performs the upgrade - `@hono/node-server` in a standalone API, or + * Nitro through `@vitnode/core/ws/crossws` in a single app. + * + * The handshake runs through the API's middleware like any request, so the + * session cookie decides the user (`null` for a guest) and the Hono context + * stays available to every handler for the connection's lifetime. With Redis + * configured it answers `503` until this instance is subscribed to the + * realtime channel, so no socket is accepted that other instances cannot reach. * * Server-only. Call it once where the API mounts WebSockets. * * @param createEvents Optional connection-level handlers. Their `onMessage` * only receives messages that matched no registered socket. + * + * @example + * ```ts + * import { upgradeWebSocket } from "@hono/node-server"; + * import { websocketOriginMiddleware } from "@vitnode/core/api/middlewares/websocket-origin.middleware"; + * import { handleVitNodeWebSocket } from "@vitnode/core/ws/handle"; + * + * app.get( + * "/ws", + * websocketOriginMiddleware(), + * upgradeWebSocket(handleVitNodeWebSocket()), + * ); + * ``` */ export function handleVitNodeWebSocket( createEvents?: (c: Context) => Promise | WSEvents, @@ -63,6 +92,11 @@ export function handleVitNodeWebSocket( markWebSocketEnabled(); return async (c: Context): Promise => { + if (!(await waitForRealtimePubSub())) { + throw new HTTPException(503, { + message: "Realtime is not ready on this instance yet.", + }); + } const inline = await createEvents?.(c); const registered = c.get("core")?.webSockets ?? []; // Authenticated from the session cookie sent with the upgrade request, so diff --git a/packages/vitnode/src/ws/instances.test.ts b/packages/vitnode/src/ws/instances.test.ts index ae338dcb2..0acee0d4d 100644 --- a/packages/vitnode/src/ws/instances.test.ts +++ b/packages/vitnode/src/ws/instances.test.ts @@ -10,6 +10,7 @@ import { onInstanceMessage, onRemoteRoomMessage, publishToInstances, + realtime, wsRegistry, } from "./registry"; @@ -46,6 +47,11 @@ const bus = () => { fromOtherInstance: (message: Record) => { deliver?.(JSON.stringify({ origin: "other-instance", ...message })); }, + /** Redis hands every subscriber the message, its own publisher included. */ + echoLast: () => { + const raw = published.at(-1); + if (raw) deliver?.(raw); + }, published: () => published.map(raw => JSON.parse(raw) as Record), }; @@ -58,14 +64,14 @@ await Promise.resolve(); const sockets: WSContext[] = []; -const socket = () => { +const socket = (userId = 1) => { const sent: unknown[] = []; const ws = { send: (raw: string) => { sent.push(JSON.parse(raw)); }, } as unknown as WSContext; - wsRegistry.add(ws, 1); + wsRegistry.add(ws, userId); sockets.push(ws); return { sent, ws }; @@ -158,3 +164,63 @@ describe("room messages from another instance", () => { expect(outside.sent).toEqual([]); }); }); + +describe("socket messages from another instance", () => { + it.each([ + [ + "a broadcast reaches every local socket", + { data: "all", id: "x", type: "broadcast" }, + [[{ data: "all", id: "x" }], [{ data: "all", id: "x" }]], + ], + [ + "a user's message reaches only that user's sockets", + { data: "one", id: "x", type: "sendToUser", userId: 2 }, + [[], [{ data: "one", id: "x" }]], + ], + [ + "a batch gives each user only their own payload", + { + data: [ + { data: "a", userId: 1 }, + { data: "b", userId: 2 }, + ], + id: "x", + type: "sendToUsers", + }, + [[{ data: "a", id: "x" }], [{ data: "b", id: "x" }]], + ], + ])("%s", (_, message, [first, second]) => { + const one = socket(1); + const two = socket(2); + + pubsub.fromOtherInstance(message); + + expect(one.sent).toEqual(first); + expect(two.sent).toEqual(second); + }); +}); + +describe("this instance's own publishes", () => { + const channel = { id: "x" }; + + it.each([ + ["broadcast", () => realtime.broadcast(channel, 1)], + ["sendToUser", () => realtime.sendToUser(1, channel, 1)], + [ + "sendToUsers", + () => realtime.sendToUsers(channel, [{ data: 1, userId: 1 }]), + ], + ["toRoom", () => realtime.toRoom("room", channel, 1)], + ])( + "reach a local socket once, though %s comes back from Redis", + (_, send) => { + const local = socket(1); + wsRegistry.join(local.ws, "room", "tab-a"); + + send(); + pubsub.echoLast(); + + expect(local.sent).toEqual([{ data: 1, id: "x" }]); + }, + ); +}); diff --git a/packages/vitnode/src/ws/manager.test.ts b/packages/vitnode/src/ws/manager.test.ts index 478bdad0e..b4265ffa3 100644 --- a/packages/vitnode/src/ws/manager.test.ts +++ b/packages/vitnode/src/ws/manager.test.ts @@ -138,3 +138,124 @@ describe("createWebSocketManager", () => { ).not.toThrow(); }); }); + +const managers: ReturnType[] = []; + +const managed = (onReadyStateChange = vi.fn()) => { + const manager = createWebSocketManager({ + onMessage: vi.fn(), + onReadyStateChange, + url: "ws://localhost/api/ws", + }); + managers.push(manager); + + return { channel: bcInstances.at(-1), manager, onReadyStateChange }; +}; + +const readyStatesPosted = (channel: FakeBroadcastChannel | undefined) => + channel?.posted.filter( + message => (message as { type: string }).type === "ready-state", + ); + +afterEach(() => { + managers.splice(0).forEach(manager => manager.destroy()); + vi.useRealTimers(); +}); + +describe("connection ids", () => { + it("are new for every socket the leader opens, and null while closed", () => { + vi.useFakeTimers(); + const { onReadyStateChange } = managed(); + + wsInstances[0]?.fireOpen(); + wsInstances[0]?.fireClose(); + vi.advanceTimersByTime(1000); + wsInstances[1]?.fireOpen(); + + const calls = onReadyStateChange.mock.calls; + expect(calls).toEqual([ + [1, expect.any(String)], + [3, null], + [1, expect.any(String)], + ]); + expect(calls[2]?.[1]).not.toBe(calls[0]?.[1]); + }); + + it("go through CONNECTING to a new one when the leader is asked to reconnect", () => { + const { channel, manager, onReadyStateChange } = managed(); + wsInstances[0]?.fireOpen(); + const first = manager.getConnectionId(); + + manager.reconnect(); + wsInstances[1]?.fireOpen(); + + expect(onReadyStateChange.mock.calls.slice(1)).toEqual([ + [0, null], + [1, manager.getConnectionId()], + ]); + expect(manager.getConnectionId()).not.toBe(first); + expect(readyStatesPosted(channel)).toEqual([ + { connectionId: first, payload: 1, type: "ready-state" }, + { connectionId: null, payload: 0, type: "ready-state" }, + { + connectionId: manager.getConnectionId(), + payload: 1, + type: "ready-state", + }, + ]); + }); + + it("are told again to a tab that asks for the state late", () => { + const { channel, manager } = managed(); + wsInstances[0]?.fireOpen(); + + channel?.onmessage?.({ data: { type: "request-state" } }); + + expect(readyStatesPosted(channel)?.at(-1)).toEqual({ + connectionId: manager.getConnectionId(), + payload: 1, + type: "ready-state", + }); + }); +}); + +describe("a follower tab", () => { + beforeEach(() => { + // Another tab holds the leader lock for as long as this test runs. + vi.stubGlobal("navigator", { + locks: { request: async () => await new Promise(() => {}) }, + }); + }); + + it("takes the leader's connection id and asks the leader to send for it", () => { + const { channel, manager, onReadyStateChange } = managed(); + + channel?.onmessage?.({ + data: { connectionId: "conn-2", payload: 1, type: "ready-state" }, + }); + manager.send({ data: 1, id: "x" }); + + expect(wsInstances).toHaveLength(0); + expect(onReadyStateChange).toHaveBeenLastCalledWith(1, "conn-2"); + expect(manager.getConnectionId()).toBe("conn-2"); + expect(channel?.posted).toContainEqual({ + payload: { data: 1, id: "x" }, + type: "send", + }); + }); +}); + +describe("the leader tab", () => { + it("sends a follower's message over its socket", () => { + const { channel } = managed(); + wsInstances[0]?.fireOpen(); + + channel?.onmessage?.({ + data: { payload: { data: 1, id: "x" }, type: "send" }, + }); + + expect(wsInstances[0]?.sent).toEqual([ + JSON.stringify({ data: 1, id: "x" }), + ]); + }); +}); diff --git a/packages/vitnode/src/ws/manager.ts b/packages/vitnode/src/ws/manager.ts index 9ec6ee17f..8e3da1dd3 100644 --- a/packages/vitnode/src/ws/manager.ts +++ b/packages/vitnode/src/ws/manager.ts @@ -7,7 +7,7 @@ import { WS_PING_ID, WS_PONG_ID } from "./types"; * {@link BroadcastChannel}. */ type CrossTabMessage = - | { payload: number; type: "ready-state" } + | { connectionId: null | string; payload: number; type: "ready-state" } | { payload: VitNodeWSMessage; type: "message" } | { payload: VitNodeWSMessage; type: "send" } | { type: "reconnect" } @@ -21,6 +21,11 @@ const RECONNECT_MAX_DELAY = 30_000; const HEARTBEAT_INTERVAL = 25_000; const HEARTBEAT_TIMEOUT = 10_000; +const newConnectionId = (): string => + typeof crypto !== "undefined" && "randomUUID" in crypto + ? crypto.randomUUID() + : `${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`; + const isOffline = (): boolean => typeof navigator !== "undefined" && "onLine" in navigator && @@ -40,13 +45,18 @@ const reconnectDelay = (attempt: number): number => { export interface WebSocketManagerOptions { /** Called once per logical server message, in every tab. */ onMessage: (message: VitNodeWSMessage) => void; - /** Called whenever the shared connection's `readyState` changes. */ - onReadyStateChange: (state: number) => void; + /** + * Called whenever the shared connection's `readyState` changes, in every + * tab. `connectionId` is new for every socket the leader opens, so a tab can + * tell a reconnect apart even when it never saw the state leave `OPEN`. + */ + onReadyStateChange: (state: number, connectionId: null | string) => void; url: string; } export interface WebSocketManager { destroy: () => void; + getConnectionId: () => null | string; getReadyState: () => number; reconnect: () => void; @@ -62,6 +72,7 @@ export const createWebSocketManager = ({ let isLeader = false; let isDestroyed = false; let readyState = 0; // WebSocket.CONNECTING + let connectionId: null | string = null; let releaseLeadership: (() => void) | undefined; let reconnectTimeout: ReturnType | undefined; let reconnectAttempt = 0; @@ -79,14 +90,15 @@ export const createWebSocketManager = ({ channel?.postMessage(message); }; - const setReadyState = (state: number) => { + const setReadyState = (state: number, id: null | string) => { readyState = state; - onReadyStateChange(state); + connectionId = id; + onReadyStateChange(state, id); }; - const publishReadyState = (state: number) => { - setReadyState(state); - post({ payload: state, type: "ready-state" }); + const publishReadyState = (state: number, id: null | string = null) => { + setReadyState(state, id); + post({ connectionId: id, payload: state, type: "ready-state" }); }; const clearPongTimeout = () => { @@ -151,7 +163,7 @@ export const createWebSocketManager = ({ ws.onopen = () => { reconnectAttempt = 0; - publishReadyState(ws.readyState); + publishReadyState(ws.readyState, newConnectionId()); outgoingQueue .splice(0) .forEach(message => ws.send(JSON.stringify(message))); @@ -231,6 +243,7 @@ export const createWebSocketManager = ({ if (!isLeader || isDestroyed) return; if (reconnectTimeout) clearTimeout(reconnectTimeout); closeSocket(); + publishReadyState(WebSocket.CONNECTING); openSocket(); }; @@ -268,13 +281,15 @@ export const createWebSocketManager = ({ onMessage(message.payload); break; case "ready-state": - if (!isLeader) setReadyState(message.payload); + if (!isLeader) setReadyState(message.payload, message.connectionId); break; case "reconnect": if (isLeader) reconnectLeader(); break; case "request-state": - if (isLeader) post({ payload: readyState, type: "ready-state" }); + if (isLeader) { + post({ connectionId, payload: readyState, type: "ready-state" }); + } break; case "send": if (isLeader) sendOverSocket(message.payload); @@ -332,6 +347,7 @@ export const createWebSocketManager = ({ return { destroy, + getConnectionId: () => connectionId, getReadyState: () => readyState, reconnect, send, diff --git a/packages/vitnode/src/ws/nitro-route.ts b/packages/vitnode/src/ws/nitro-route.ts new file mode 100644 index 000000000..a28f77ecd --- /dev/null +++ b/packages/vitnode/src/ws/nitro-route.ts @@ -0,0 +1,35 @@ +import type { EventHandler } from "nitro/h3"; + +import { defineWebSocketHandler } from "nitro"; +import { fetchViteEnv } from "nitro/vite/runtime"; + +const warmup = async (request: Request) => fetchViteEnv("ssr", request); + +const loadHooks = async (url: string) => { + await warmup(new Request(url)); + const { createVitNodeWebSocketHooks } = await import("./crossws.server"); + + return createVitNodeWebSocketHooks({ warmup }); +}; + +let hooks: ReturnType | undefined; + +const vitNodeWebSocketRoute: EventHandler = defineWebSocketHandler( + async event => { + hooks ??= loadHooks(event.req.url); + try { + return await hooks; + } catch (error) { + hooks = undefined; + // oxlint-disable-next-line no-console + console.error("[VitNode] The WebSocket route could not load:", error); + + return { + upgrade: () => + new Response("WebSocket is unavailable.", { status: 503 }), + }; + } + }, +); + +export default vitNodeWebSocketRoute; diff --git a/packages/vitnode/src/ws/provider.tsx b/packages/vitnode/src/ws/provider.tsx index 8ea725327..200147081 100644 --- a/packages/vitnode/src/ws/provider.tsx +++ b/packages/vitnode/src/ws/provider.tsx @@ -10,6 +10,7 @@ import { createWebSocketManager } from "./manager"; type Listener = (data: unknown) => void; interface VitNodeWebSocketContextValue { + connectionId: null | string; readyState: number; reconnect: () => void; send: (id: string, data: unknown) => void; @@ -37,17 +38,22 @@ const getWebSocketUrl = (): string => { return url.toString(); }; +// A separate API always serves `/api/ws`; an app that embeds the API serves it +// only when its Vite config enables `webSocket` (and mounts the Nitro route). const shouldConnectWebSocket = (): boolean => + process.env.VITNODE_EMBEDDED_WEBSOCKET === "true" || CONFIG.api.pathname !== CONFIG.web.pathname || CONFIG.api.port !== CONFIG.web.port; /** * Owns the browser's single WebSocket connection to `/api/ws` for * `useVitNodeWebSocket` hooks below it. Tabs share one socket - * (one tab holds it and relays for the others) and it reconnects after a drop. + * (one tab holds it and relays for the others) and it reconnects after a drop, + * with backoff and jitter - expected on serverless hosts, which close sockets + * at the function's maximum duration. * - * It does not connect when the API is served from the same path and port as - * the web app. + * In an app that embeds the API it connects only when `vitnode({ webSocket: + * true })` is set in `vite.config.ts`; with a separate API it always does. */ export const VitNodeWebSocketProvider = ({ children, @@ -57,10 +63,20 @@ export const VitNodeWebSocketProvider = ({ const listenersRef = React.useRef>>(new Map()); const managerRef = React.useRef(null); const pendingSendRef = React.useRef([]); - const [readyState, setReadyState] = React.useState(() => { - if (typeof WebSocket === "undefined") return 0; + const [connection, setConnection] = React.useState<{ + connectionId: null | string; + readyState: number; + }>(() => { + if (typeof WebSocket === "undefined") { + return { connectionId: null, readyState: 0 }; + } - return shouldConnectWebSocket() ? WebSocket.CONNECTING : WebSocket.CLOSED; + return { + connectionId: null, + readyState: shouldConnectWebSocket() + ? WebSocket.CONNECTING + : WebSocket.CLOSED, + }; }); const dispatch = React.useCallback((message: VitNodeWSMessage) => { @@ -75,9 +91,9 @@ export const VitNodeWebSocketProvider = ({ const manager = createWebSocketManager({ url: getWebSocketUrl(), onMessage: dispatch, - onReadyStateChange: state => { + onReadyStateChange: (readyState, connectionId) => { React.startTransition(() => { - setReadyState(state); + setConnection({ connectionId, readyState }); }); }, }); @@ -118,9 +134,10 @@ export const VitNodeWebSocketProvider = ({ managerRef.current?.reconnect(); }, []); + const { connectionId, readyState } = connection; const value = React.useMemo( - () => ({ readyState, reconnect, send, subscribe }), - [readyState, reconnect, send, subscribe], + () => ({ connectionId, readyState, reconnect, send, subscribe }), + [connectionId, readyState, reconnect, send, subscribe], ); return ( diff --git a/packages/vitnode/src/ws/realtime-readiness.test.ts b/packages/vitnode/src/ws/realtime-readiness.test.ts new file mode 100644 index 000000000..f279380a6 --- /dev/null +++ b/packages/vitnode/src/ws/realtime-readiness.test.ts @@ -0,0 +1,244 @@ +import type { WSContext } from "hono/ws"; + +import { Hono } from "hono"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import type { CacheClient } from "@/api/lib/cache"; +import type { EnvVitNode } from "@/api/middlewares/global.middleware"; + +type SubscriberEvent = "end" | "error" | "ready" | "reconnecting"; + +/** + * A Redis client whose subscriber connection the test drives: `subscribe` + * answers from `subscribeResults` in turn (pending forever once they run out) + * and `emit` fires what node-redis would. + */ +const fakeRedis = ({ + connect = async () => await Promise.resolve(), + subscribeResults = [] as (() => Promise)[], +} = {}) => { + const listeners = new Map void)[]>(); + const published: Promise[] = []; + const subscribe = vi.fn( + async () => + await (subscribeResults.shift() ?? (async () => new Promise(() => {})))(), + ); + + const subscriber = { + connect, + on: (event: SubscriberEvent, listener: () => void) => { + listeners.set(event, [...(listeners.get(event) ?? []), listener]); + + return subscriber; + }, + subscribe, + }; + + const client = { + duplicate: () => subscriber, + publish: async () => { + const task = Promise.resolve(1); + published.push(task); + + return await task; + }, + }; + + return { + client: client as unknown as CacheClient, + emit: (event: SubscriberEvent) => { + listeners.get(event)?.forEach(listener => { + listener(); + }); + }, + subscribe, + }; +}; + +/** `initRealtimePubSub` runs once per module instance, so each test gets its own. */ +const freshRegistry = async () => { + vi.resetModules(); + + return await import("./registry"); +}; + +const flush = async () => { + for (let tick = 0; tick < 5; tick++) await Promise.resolve(); +}; + +const settled = (promise: Promise) => { + const state: { value?: T } = {}; + void promise.then(value => { + state.value = value; + }); + + return state; +}; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("waitForRealtimePubSub", () => { + it("is ready at once without Redis", async () => { + const { waitForRealtimePubSub } = await freshRegistry(); + + await expect(waitForRealtimePubSub()).resolves.toBe(true); + }); + + it("waits until this instance is subscribed", async () => { + let subscribed!: () => void; + const redis = fakeRedis({ + subscribeResults: [ + async () => + await new Promise(resolve => { + subscribed = resolve; + }), + ], + }); + const { initRealtimePubSub, waitForRealtimePubSub } = await freshRegistry(); + initRealtimePubSub(redis.client); + + const ready = settled(waitForRealtimePubSub()); + await flush(); + expect(ready.value).toBeUndefined(); + + subscribed(); + await flush(); + + expect(ready.value).toBe(true); + }); + + it("gives up after the timeout when the subscription never lands", async () => { + vi.useFakeTimers(); + const redis = fakeRedis(); + const { initRealtimePubSub, waitForRealtimePubSub } = await freshRegistry(); + initRealtimePubSub(redis.client); + + const ready = waitForRealtimePubSub(1000); + await vi.advanceTimersByTimeAsync(1000); + + await expect(ready).resolves.toBe(false); + }); +}); + +describe("recovering the subscription", () => { + it.each([ + [ + "the first connect fails", + { + connect: async () => await Promise.reject(new Error("ECONNREFUSED")), + subscribeResults: [async () => await Promise.resolve()], + }, + ], + [ + "the first subscribe fails", + { + subscribeResults: [ + async () => await Promise.reject(new Error("LOADING")), + async () => await Promise.resolve(), + ], + }, + ], + ])("subscribes on the next ready when %s", async (_, options) => { + const redis = fakeRedis(options); + const { initRealtimePubSub, waitForRealtimePubSub } = await freshRegistry(); + initRealtimePubSub(redis.client); + await flush(); + const ready = settled(waitForRealtimePubSub()); + await flush(); + expect(ready.value).toBeUndefined(); + + redis.emit("ready"); + await flush(); + + expect(ready.value).toBe(true); + }); + + it.each(["reconnecting", "end"] as const)( + "holds handshakes after %s and reconnects local sockets once ready again", + async event => { + vi.spyOn(console, "warn").mockImplementation(() => undefined); + const redis = fakeRedis({ + subscribeResults: [async () => await Promise.resolve()], + }); + const { initRealtimePubSub, waitForRealtimePubSub, wsRegistry } = + await freshRegistry(); + initRealtimePubSub(redis.client); + await flush(); + const close = vi.fn(); + wsRegistry.add({ close, send: vi.fn() } as unknown as WSContext, null); + + redis.emit("ready"); + expect(close).not.toHaveBeenCalled(); + + redis.emit(event); + const ready = settled(waitForRealtimePubSub()); + await flush(); + expect(ready.value).toBeUndefined(); + + redis.emit("ready"); + await flush(); + + expect(ready.value).toBe(true); + expect(close).toHaveBeenCalledTimes(1); + expect(close).toHaveBeenCalledWith(1012, "Realtime resync"); + expect(redis.subscribe).toHaveBeenCalledTimes(1); + }, + ); +}); + +describe("the WebSocket handshake", () => { + it("answers 503 while this instance is not subscribed", async () => { + vi.useFakeTimers(); + const { initRealtimePubSub } = await freshRegistry(); + const { handleVitNodeWebSocket } = await import("./handle"); + const { createVitNodeWebSocketHooks, upgradeWebSocket } = + await import("./crossws.server"); + initRealtimePubSub(fakeRedis().client); + const app = new Hono().basePath("/api"); + app.get("/ws", upgradeWebSocket(handleVitNodeWebSocket())); + const hooks = createVitNodeWebSocketHooks({ + api: async request => await app.fetch(request), + }); + + const rejection = hooks + .upgrade(new Request("https://site.example/api/ws")) + .catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(5000); + + const response = await rejection; + expect(response).toBeInstanceOf(Response); + expect((response as Response).status).toBe(503); + }); +}); + +describe("runWithRealtimeKeepAlive", () => { + it("hands over every publish started inside it, and none outside", async () => { + const redis = fakeRedis({ + subscribeResults: [async () => await Promise.resolve()], + }); + const { + initRealtimePubSub, + publishToInstances, + realtime, + runWithRealtimeKeepAlive, + } = await freshRegistry(); + initRealtimePubSub(redis.client); + const keepAlive = vi.fn(); + const channel = { id: "test" }; + + await runWithRealtimeKeepAlive(keepAlive, async () => { + realtime.broadcast(channel, 1); + await Promise.resolve(); + realtime.sendToUser(1, channel, 2); + publishToInstances("presence", 3); + }); + realtime.broadcast(channel, 4); + + expect(keepAlive).toHaveBeenCalledTimes(3); + for (const [task] of keepAlive.mock.calls) { + expect(task).toBeInstanceOf(Promise); + } + }); +}); diff --git a/packages/vitnode/src/ws/registry.ts b/packages/vitnode/src/ws/registry.ts index 0247cf25e..8f1ea709b 100644 --- a/packages/vitnode/src/ws/registry.ts +++ b/packages/vitnode/src/ws/registry.ts @@ -1,5 +1,7 @@ import type { WSContext } from "hono/ws"; +import { AsyncLocalStorage } from "node:async_hooks"; + import type { CacheClient } from "@/api/lib/cache"; import type { VitNodeWSChannel, VitNodeWSMessage } from "./types"; @@ -156,6 +158,20 @@ export const wsRegistry = { } }); }, + /** + * Closes every connection on this instance with `code`, so each client + * reconnects and restores its rooms and state. Cleanup runs through each + * socket's normal close handling. + */ + closeAll: (code: number, reason: string): void => { + for (const ws of [...connections.keys()]) { + try { + ws.close(code, reason); + } catch { + wsRegistry.remove(ws); + } + } + }, /** The rooms one socket is in. */ roomsOf: (ws: WSContext): string[] => [...(roomsBySocket.get(ws) ?? [])], /** Send `{ id, data }` to every socket in `room` on this instance. */ @@ -256,10 +272,26 @@ export const onInstanceMessage = ( export const getRealtimeInstanceId = (): string => instanceId; +type KeepAlive = (task: Promise) => void; + +// The request a publish happens in, so a serverless runtime keeps the instance +// alive until Redis has the message instead of freezing it with the response. +const keepAliveScope = new AsyncLocalStorage(); + +/** + * Runs `fn` with `keepAlive` receiving every realtime publish started inside + * it, including ones from code that has no request context of its own. The + * global middleware passes the request's `waitUntil`. Server-only. + */ +export const runWithRealtimeKeepAlive = ( + keepAlive: KeepAlive, + fn: () => T, +): T => keepAliveScope.run(keepAlive, fn); + const publish = (message: Omit): void => { if (!publisher) return; - void publisher + const task = publisher .publish( REALTIME_PUBSUB_CHANNEL, JSON.stringify({ ...message, origin: instanceId }), @@ -268,6 +300,57 @@ const publish = (message: Omit): void => { // Other instances miss this message, but the local clients were already // served directly, so realtime keeps working on this instance. }); + keepAliveScope.getStore()?.(task); +}; + +type PubSubState = "connecting" | "down" | "ready"; + +let pubSubState: PubSubState = "connecting"; +const readyWaiters = new Set<() => void>(); + +const setPubSubState = (state: PubSubState): void => { + pubSubState = state; + if (state !== "ready") return; + readyWaiters.forEach(resolve => resolve()); + readyWaiters.clear(); +}; + +/** + * Pub/Sub has no replay, so a socket that stayed open while this instance was + * unsubscribed may have missed messages. Closing it makes the client reconnect + * and run its normal restore: rejoin rooms and reload durable state. + */ +const resyncLocalConnections = (): void => { + // oxlint-disable-next-line no-console + console.warn( + "[VitNode] Realtime pub/sub resubscribed after a Redis outage; reconnecting local WebSocket clients so they reload missed state.", + ); + wsRegistry.closeAll(1012, "Realtime resync"); +}; + +/** + * Resolves `true` once this instance can be reached by every other instance: + * subscribed to the Redis channel, or running without Redis at all (local + * delivery only). Resolves `false` when that does not happen within + * `timeoutMs` - the WebSocket handshake answers `503` then, and the client + * retries with backoff. Server-only. + */ +export const waitForRealtimePubSub = async ( + timeoutMs = 5000, +): Promise => { + if (!publisher || pubSubState === "ready") return true; + + return new Promise(resolve => { + const done = () => { + clearTimeout(timer); + resolve(true); + }; + const timer = setTimeout(() => { + readyWaiters.delete(done); + resolve(false); + }, timeoutMs); + readyWaiters.add(done); + }); }; export const initRealtimePubSub = (client: CacheClient | null): void => { @@ -279,40 +362,81 @@ export const initRealtimePubSub = (client: CacheClient | null): void => { // A connection in subscribe mode can't run other commands, so use a // dedicated duplicate for receiving. `duplicate()` returns a disconnected // client, hence the explicit `connect()`; node-redis restores the - // subscription itself after every reconnect, so this only runs once. + // subscription itself after every reconnect, so it is only made once. const subscriber = client.duplicate(); + let subscribed = false; + let subscribing = false; + + const onMessage = (raw: string): void => { + try { + const message = JSON.parse(raw) as RealtimePubSubMessage; + // Skip our own echo - the publisher already delivered locally. + if (message.origin === instanceId) return; + if (message.type === "instance") { + const { data, id: topic, origin } = message; + instanceListeners.forEach(listener => { + listener({ data, origin, topic }); + }); + + return; + } + if (message.type === "room" && typeof message.room === "string") { + const { data, id, room } = message; + roomListeners.forEach(listener => listener(room, id, data)); + } + deliverLocally(message); + } catch { + // Ignore malformed payloads. + } + }; + + // Until this succeeds the instance is not ready: WebSocket handshakes wait + // for it, so no socket is accepted that other instances cannot reach. A + // failed attempt is retried on the next `ready`. + const subscribe = (): void => { + if (subscribed || subscribing) return; + subscribing = true; + subscriber + .subscribe(REALTIME_PUBSUB_CHANNEL, onMessage) + .then(() => { + subscribed = true; + setPubSubState("ready"); + }) + .catch(() => { + // Realtime stays single-instance; local clients still get their messages. + }) + .finally(() => { + subscribing = false; + }); + }; + subscriber.on("error", () => { // Connection errors are non-fatal; local delivery still works. }); + // node-redis resubscribes before it emits `ready` again, so a `ready` after a + // drop means messages flow again - but the ones sent meanwhile are gone. + subscriber.on("reconnecting", () => { + if (subscribed) setPubSubState("down"); + }); + subscriber.on("end", () => { + if (subscribed) setPubSubState("down"); + }); + subscriber.on("ready", () => { + if (!subscribed) { + subscribe(); + + return; + } + if (pubSubState !== "down") return; + setPubSubState("ready"); + resyncLocalConnections(); + }); void subscriber .connect() - .then(async () => - subscriber.subscribe(REALTIME_PUBSUB_CHANNEL, raw => { - try { - const message = JSON.parse(raw) as RealtimePubSubMessage; - // Skip our own echo - the publisher already delivered locally. - if (message.origin === instanceId) return; - if (message.type === "instance") { - const { data, id: topic, origin } = message; - instanceListeners.forEach(listener => { - listener({ data, origin, topic }); - }); - - return; - } - if (message.type === "room" && typeof message.room === "string") { - const { data, id, room } = message; - roomListeners.forEach(listener => listener(room, id, data)); - } - deliverLocally(message); - } catch { - // Ignore malformed payloads. - } - }), - ) + .then(subscribe) .catch(() => { - // Realtime stays single-instance; local clients still get their messages. + // The reconnect strategy keeps retrying; `ready` subscribes then. }); }; @@ -325,8 +449,11 @@ export const publishToInstances = (topic: string, data: unknown): void => { * `c.get("realtime")` in API handlers. * * Delivery is best effort and fire-and-forget: clients that are not connected - * miss the message. Without Redis (`REDIS_URL`) only clients connected to the - * current instance receive it; with Redis it is relayed to every instance. + * miss the message, and nothing is replayed after a reconnect - keep durable + * state in the database and let clients reload it when they reconnect. Without + * Redis (`REDIS_URL`) only clients connected to the current instance receive + * it; with Redis it is relayed to every instance, and the publish is kept + * alive past the HTTP response with the request's `waitUntil`. */ export interface VitNodeRealtime { /** Push a payload to every open connection. */ diff --git a/packages/vitnode/src/ws/use-websocket.ts b/packages/vitnode/src/ws/use-websocket.ts index 198236d2b..b1f60780b 100644 --- a/packages/vitnode/src/ws/use-websocket.ts +++ b/packages/vitnode/src/ws/use-websocket.ts @@ -5,6 +5,13 @@ import type { VitNodeWSChannel } from "./types"; import { useVitNodeWebSocketContext } from "./provider"; export interface UseVitNodeWebSocketReturn { + /** + * Changes every time the shared connection (re)opens, in every tab, and is + * `null` while it is not open. Depend on it to restore server-side state + * after a reconnect - rejoin rooms, resubscribe, reload what may have been + * missed. The server keeps no state for a closed socket and replays nothing. + */ + connectionId: null | string; /** * The current `readyState` of the shared connection (`0` CONNECTING, `1` * OPEN, `2` CLOSING, `3` CLOSED). @@ -20,6 +27,9 @@ export interface UseVitNodeWebSocketReturn { * the tree and throws without it. * * Messages sent before the connection opens are queued by the provider. + * Delivery is best effort: a reconnect - after a network drop, a deploy, or a + * serverless host's duration limit - loses rooms and anything pushed meanwhile, + * so restore them when `connectionId` changes. * * @param channel A channel from `createWebSocketChannel`, or a raw socket id. * @param options.onMessage Called with each payload the server sends to this @@ -27,9 +37,13 @@ export interface UseVitNodeWebSocketReturn { * * @example * ```tsx - * const { send } = useVitNodeWebSocket(echoChannel, { + * const { connectionId, send } = useVitNodeWebSocket(roomChannel, { * onMessage: data => setText(data.text), * }); + * + * React.useEffect(() => { + * if (connectionId) send({ room: "lobby", type: "join" }); + * }, [connectionId, send]); * ``` */ export function useVitNodeWebSocket( @@ -39,6 +53,7 @@ export function useVitNodeWebSocket( }, ): UseVitNodeWebSocketReturn { const { + connectionId, readyState, send: contextSend, subscribe, @@ -61,5 +76,9 @@ export function useVitNodeWebSocket( [contextSend, id], ); - return { readyState, send }; + return { + connectionId: readyState === 1 ? connectionId : null, + readyState, + send, + }; } diff --git a/packages/vitnode/vitest.config.ts b/packages/vitnode/vitest.config.ts index b2739bb89..4f50b932d 100644 --- a/packages/vitnode/vitest.config.ts +++ b/packages/vitnode/vitest.config.ts @@ -26,7 +26,9 @@ const ISOLATED = [ "src/tanstack/admin/content/slots.test.ts", "src/tanstack/admin/transport.test.ts", "src/tanstack/auth/transport.test.ts", + "src/ws/crossws.server.test.ts", "src/ws/pubsub-lifetime.test.ts", + "src/ws/realtime-readiness.test.ts", ]; // Real builds, watchers and child processes. `pnpm test` still runs them; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index fb06dc916..9cfe6019a 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -781,6 +781,9 @@ importers: lucide-react: specifier: ^1.53.0 version: 1.53.0(@types/react@19.3.0)(react@19.3.0) + nitro: + specifier: 3.0.260903-beta + version: 3.0.260903-beta oxlint: specifier: ^1.87.0 version: 1.87.0(oxlint-tsgolint@7.0.2003)