diff --git a/docs/guide/websockets.md b/docs/guide/websockets.md index 1b06b18ab..40620cfe8 100644 --- a/docs/guide/websockets.md +++ b/docs/guide/websockets.md @@ -91,6 +91,8 @@ export default defineRoom({ - `ctx.connection.close(code?, reason?)` - `ctx.connection.setState(data)` +Add `actions` when other server code should reach the room. See [Server-side actions](#server-side-actions). + ## `defineWebSocket()` Use `defineWebSocket()` when each connection is handled independently instead of as a shared room. @@ -156,7 +158,76 @@ When Void auth is enabled, every WebSocket hook receives the current user as `ct Connections from another origin are anonymous unless that origin is in your auth `trustedOrigins`. If you read your own cookies from `ctx.request` in `onBeforeConnect`, check its `Origin` header yourself. -`ctx.user` is resolved when the socket connects and stays the same for that connection, even if the user later signs out or the session is revoked. For actions that must respect a ban or role change, re-check your own records in `onMessage` and close the connection with `ctx.connection.close()` in rooms or `ctx.socket.close()` in `defineWebSocket()`. +`ctx.user` is resolved when the socket connects and stays the same for that connection, even if the user later signs out or the session is revoked. For actions that must respect a ban or role change, re-check your own records in `onMessage` and close the connection with `ctx.connection.close()` in rooms or `ctx.socket.close()` in `defineWebSocket()`. To close idle sockets as soon as a session is revoked, declare a [room action](#server-side-actions) and call it from the code that revokes the session. + +## Server-side actions + +A room can declare named actions that other server code calls with typed arguments. An action runs inside the room's Durable Object, so it can close connections or broadcast without waiting for a client frame. + +```ts +// routes/relay/[accountId].ws.ts +import * as v from 'valibot'; +import { getSession } from 'void/auth'; +import { defineRoom } from 'void/ws'; + +const ClientMessage = v.object({ type: v.literal('relay.ping') }); +const ServerMessage = v.object({ type: v.literal('relay.message'), text: v.string() }); + +export default defineRoom({ + messages: { client: ClientMessage, server: ServerMessage }, + onConnect(ctx) { + // The id comes from the authenticated session, never from client input. + const session = getSession()?.session; + if (session) { + ctx.connection.setState({ sessionId: session.id }); + } + }, + actions: { + async revokeSession(ctx, sessionId: string, code: number) { + let closed = 0; + for (const connection of ctx.room.getConnections()) { + const state = connection.state as { sessionId?: string } | undefined; + if (state?.sessionId !== sessionId) continue; + connection.close(code, 'session revoked'); + closed += 1; + } + return { closed }; + }, + async announce(ctx, text: string) { + await ctx.room.broadcast({ type: 'relay.message', text }); + }, + }, +}); +``` + +Call an action from a route handler, an auth hook, a queue consumer or a cron job with `getRoom()` from `void/ws/server`: + +```ts +import { getRoom } from 'void/ws/server'; + +// Better Auth's databaseHooks.session.delete.after, or any other server code +const room = getRoom('/relay/[accountId]', { params: { accountId } }); +const { closed } = await room.actions.revokeSession(session.id, 4001); +``` + +`getRoom()` addresses the same instance `connect()` reaches. Pass the route pattern and the same params, and a custom `key` is applied too. + +Match connections on server-verified state. A session id read from the URL, a query string or a client message can be any value a client sends, so the revocation would close the wrong sockets. Derive the id from `getSession()` or your own records, store it in `onConnect`, and pass that same id from the code that revokes it. + +- Annotate each action parameter. Arguments and results are typed from the definition, and an unannotated parameter is `never`. +- Actions receive `ctx.id`, `ctx.params`, `ctx.env`, `ctx.storage` and `ctx.room`. There is no `ctx.request` or `ctx.user`: other server code is the caller, so authorize the decision where you make it. +- `ctx.room.broadcast()` and connection `send()` accept any event inside an action. The server message schema still validates it at runtime. +- Arguments cross a Durable Object boundary, so pass structured-clone data, not callbacks or class instances. +- An unknown action name or a handler error rejects the call. An unknown route, a `defineWebSocket()` route or a missing param throws before any Durable Object is contacted. +- Do not call an action from the same room's own hooks. Use `ctx.room` there; a call from inside the room waits on the room itself. + +Actions are not an endpoint. They run over Durable Object RPC from your Worker, so a browser request or a WebSocket message cannot call one. + +`getRoom()` reads the current runtime env, so it works in any handler Void runs. Outside one, pass the bindings object you already hold: + +```ts +getRoom('/relay/[accountId]', { params: { accountId }, env: bindings }); +``` ## Hooks diff --git a/docs/reference/api.md b/docs/reference/api.md index b5460f654..dafda33b1 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -42,7 +42,8 @@ Import Void's app-facing APIs from these paths: | `void/auth` | `defineAuth`, `getUser`, `getSession`, `requireAuth`, `AuthUser`, `AuthSession`, `AuthState` | | `void/client` | `fetch`, `fetchStream`, `FetchError`, `auth`, `createAuthClient`, `AuthUser`, `AuthSession`, `AuthState` | | `void/client/{framework}` | Same as `void/client`, with framework-specific Better Auth clients for `react`, `vue`, `svelte`, and `solid` | -| `void/ws` | `defineRoom`, `defineWebSocket`, `connect`, WebSocket context and connection types | +| `void/ws` | `defineRoom`, `defineWebSocket`, `connect`, WebSocket context, connection and action types | +| `void/ws/server` | `getRoom`, `RoomHandle`, `RoomOptions`, `RoomPath`. Server-only: call room actions from routes, auth hooks, queues and cron jobs. | | `void/durable` | `defineDurableState`, `defineDurableObject`, typed state/RPC/context/migration types | | `void/sse` | `eventStream`, `formatSse`, `formatSseText`, `getLastEventId`, `SseStreamClosedError`, SSE message and stream types | | `void/sse/client` | `connectEventStream`, browser `EventSource` wrapper types | diff --git a/packages/void/AGENTS.md b/packages/void/AGENTS.md index 3842f09dc..29a3ec54c 100644 --- a/packages/void/AGENTS.md +++ b/packages/void/AGENTS.md @@ -131,6 +131,7 @@ generated entries must preserve the module-boundary rules below. - **Reserved `__VOID_*` names**: `envSchemaPlugin.config` (every mode, with or without `env.ts`) and `void env check` reject `__VOID_*` names in the root `.env` or declared in `env.ts`. Do not try to strip them from dev `vars` instead: Wrangler reads `.env` itself and writes it over `vars`, so a strip cannot be enforced. Manual trigger auth is a compile-time literal (`__VOID_SERVE` from `CompileOptions.dev`, false in builds and `vite preview`): dev accepts only `x-void-dev-trigger`, builds use the proxy token or build key. Never key it on `import.meta.env.DEV`, which is true in a `NODE_ENV=development` build. - **Typed Durable State convention**: each default-exported `defineDurableState()` definition in `durable-objects/*.ts` becomes a filename-derived binding and worker class (`shopping-cart.ts` → `SHOPPING_CART` / `ShoppingCartDurableObject`). The generated entry calls the internal `__bind()` hook so named imports can use `State.get(name)` without touching `c.env`, exports `State.class`, and emits a stable `new_sqlite_classes` migration. `syncWranglerDurableObjects()` persists bindings and append-only class migration history in `void.lock.json` through the generated Cloudflare config; never regenerate migration order solely from the current sorted files. The generated class extends `DurableObject` from `cloudflare:workers` so namespace stubs expose real RPC methods; runtime state and helpers use JavaScript private fields and methods. Runtime operations are serialized and persisted after successful typed RPC, `fetch`, or `alarm` handlers; versioned value migrations run on load. This is native-Void + Cloudflare only and is supported by direct Cloudflare deploys. Typed `*.ws.ts` routes likewise use SQLite-backed classes; their root-config sync uses the comment-preserving writer, retains legacy `new_classes` histories, and appends a new migration tag only for new classes. - **Persistent resource names**: optional static `name` on Durable State and WebSocket definitions uses the same identity derivation as unnamed filenames/routes. Preserve unnamed defaults exactly. `void info` reports names and verified unmatched lock identities without executing source or writing files. Scanners use the shared AST reader; runtime validation imports only the Worker-safe `shared/resource-name.ts` leaf. Carry explicit names through generated metadata and `__bind()`; runtime mismatches fail instead of selecting an inferred replacement. WebSocket `key({ params })` is shared by namespace selection and hook `ctx.id`; absent keys keep the existing ordered `param=value` format. +- **Server room actions**: `defineRoom({ actions })` keeps each action's declared parameters and return type (`RoomActions` accepts any annotated parameter list, so unannotated parameters become `never`). Codegen adds `actions` to `WebSocketRouteMap` for room routes only, and `getRoom()` from `void/ws/server` resolves the same `bindingName` plus `buildWebSocketInstanceId()` key as `forwardWebSocketRequest()`. The generated route table registers route metadata once, after attaching `key`, through `__registerWebSocketRoutes`; the map lives on `globalThis` because dev can evaluate the module twice. Dispatch is Durable Object RPC (`__voidRoomAction`), so the WebSocket DO class extends `DurableObject` from `cloudflare:workers`; its own helpers are `#` methods so they stay off that RPC surface, and actions are never reachable from `fetch`, `webSocketMessage`, or client HTTP. - **Journal coherence invariant**: `db/migrations/*.sql` and `db/migrations/meta/_journal.json` must stay in sync — every `.sql` file has a matching journal entry (by `tag`) and vice versa. `assertJournalCoherence` in `src/migrations/validate.ts` is the gate, called from `deploy`, `db status`, `db migrate`, and `db reset`. The two CLI commands that mutate migration state — `void gen migration` and `void db rename-migrations` — are responsible for keeping the journal coherent themselves: when the journal has entries, the first delegates to `drizzle-kit generate --custom --name `, which writes the empty `.sql`, appends the journal entry, and writes a snapshot of the current schema (it refuses if the last entry's snapshot is missing and byte-restores the journal if drizzle-kit fails); with no journal entries it writes a plain timestamped `.sql`. The second rewrites `tag` values alongside the file renames and also renames the matching `meta/_snapshot.json` (with byte-level rollback on failure). Drift will fail loudly before touching any remote DB. This prevents an orphan scaffolded `.sql` file from half-applying against D1 and blocking later deploys. - **D1 migration recovery**: Publish the lock and attempt record atomically. Stable attempts register in `_void_migration_fences`; every migration and legacy tracking-table upgrade checks ownership inside its D1 batch. Recovery revokes ownership before reading confirmed migrations, so a resumed executor cannot write after a replacement starts. Duplicate submissions only report progress. Never steal a missing-record lock or interrupt an older/raw-exec attempt without fencing: its owner may still be running. Such interrupted legacy attempts require operator verification that the executor has stopped before manually recovering the database lock. diff --git a/packages/void/package.json b/packages/void/package.json index 0b91cb170..08fe45c1e 100644 --- a/packages/void/package.json +++ b/packages/void/package.json @@ -154,6 +154,11 @@ "import": "./dist/runtime/ws.mjs", "require": "./dist/runtime/ws.mjs" }, + "./ws/server": { + "types": "./dist/runtime/ws-room.d.mts", + "import": "./dist/runtime/ws-room.mjs", + "require": "./dist/runtime/ws-room.mjs" + }, "./db": { "types": "./dist/runtime/db.d.mts", "import": "./dist/runtime/db.mjs", diff --git a/packages/void/src/codegen/route-types.ts b/packages/void/src/codegen/route-types.ts index 682b8b897..c420eb62a 100644 --- a/packages/void/src/codegen/route-types.ts +++ b/packages/void/src/codegen/route-types.ts @@ -384,6 +384,14 @@ export function generateRouteTypes( lines.push( ` server: ${imports.map((source) => `ExtractOutput<${source}["server"]>`).join(' | ')};`, ); + const roomSources = sources.filter((route) => route.mode === 'room'); + if (roomSources.length > 0) { + const actions = roomSources.map((route) => { + const importPath = relative(outDir, `${routesDir}/${route.filePath}`); + return `NonNullable<(typeof import(${JSON.stringify(importPath)}))["default"]["actions"]>`; + }); + lines.push(` actions: ${actions.join(' | ')};`); + } lines.push(' };'); legacyRouteAlias(lines, 'WebSocketRouteMap', pattern); } diff --git a/packages/void/src/router/compile.ts b/packages/void/src/router/compile.ts index 65e4033b5..351dfb202 100644 --- a/packages/void/src/router/compile.ts +++ b/packages/void/src/router/compile.ts @@ -223,7 +223,7 @@ export function compileRouteTable( options.webSocketRuntimePath ) { lines.push( - `import { createWebSocketDurableObject as __createWebSocketDurableObject, createWebSocketTombstoneDurableObject as __createWebSocketTombstoneDurableObject, forwardWebSocketRequest as __forwardWebSocketRequest } from ${JSON.stringify(options.webSocketRuntimePath)};`, + `import { createWebSocketDurableObject as __createWebSocketDurableObject, createWebSocketTombstoneDurableObject as __createWebSocketTombstoneDurableObject, forwardWebSocketRequest as __forwardWebSocketRequest, __registerWebSocketRoutes } from ${JSON.stringify(options.webSocketRuntimePath)};`, ); } if (options.hasLive && options.liveRuntimePath) { @@ -409,6 +409,7 @@ export function compileRouteTable( params: route.params, className: route.className, bindingName: route.bindingName, + mode: route.mode, ...(route.name !== undefined && { name: route.name }), }, )}, ${auth ? '__voidAuthOptions' : 'undefined'}, ${auth ? '__resolveVoidAuthState' : 'undefined'});`, @@ -422,6 +423,7 @@ export function compileRouteTable( params: route.params, className: route.className, bindingName: route.bindingName, + mode: route.mode, ...(route.name !== undefined && { name: route.name }), })), )};`, @@ -431,6 +433,10 @@ export function compileRouteTable( for (let i = 0; i < webSockets.length; i++) { lines.push(`__voidWebSocketRoutes[${i}].key = websocket${i}.key;`); } + if (webSockets.length > 0 && options.webSocketRuntimePath) { + // Register after `key` is attached so getRoom() shares the client's addressing. + lines.push('__registerWebSocketRoutes(__voidWebSocketRoutes);'); + } for (const className of options.webSocketTombstones ?? []) { lines.push( `export const ${className} = __createWebSocketTombstoneDurableObject(${JSON.stringify(className)});`, diff --git a/packages/void/src/runtime/ws-room.ts b/packages/void/src/runtime/ws-room.ts new file mode 100644 index 000000000..df7536a8d --- /dev/null +++ b/packages/void/src/runtime/ws-room.ts @@ -0,0 +1,148 @@ +/** + * Server-side room handle — exposed via `void/ws/server`. + * + * `getRoom()` addresses the same Durable Object instance a client's `connect()` + * reaches and calls the actions declared by `defineRoom()`. Actions run inside the + * room's Durable Object, so they can broadcast or close connections locally. + */ +import type { WebSocketRouteMap } from 'void/routes'; +import { getRawRuntimeEnv, type RuntimeEnv } from './env-raw.ts'; +import { + buildWebSocketInstanceId, + __getWebSocketRoute, + type WebSocketRouteMeta, +} from './ws-server.ts'; +import type { RoomActionContext } from './ws.ts'; + +type RouteParamsOf

= WebSocketRouteMap[P] extends { + params: infer T; +} + ? T + : Record; + +type HasParams

= + RouteParamsOf

extends Record + ? false + : [keyof RouteParamsOf

] extends [never] + ? false + : true; + +type RoomActionsOf

= WebSocketRouteMap[P] extends { + actions: infer T; +} + ? T + : never; + +/** Route patterns that declare `actions` in `defineRoom()`. */ +export type RoomPath = { + [P in keyof WebSocketRouteMap]: WebSocketRouteMap[P] extends { actions: unknown } ? P : never; +}[keyof WebSocketRouteMap]; + +type ActionArgs = T extends (context: RoomActionContext, ...args: infer A) => unknown + ? A + : never; + +type ActionResult = T extends (context: RoomActionContext, ...args: Array) => infer R + ? Awaited + : never; + +/** Handle for one room instance. Each action is called like a local async function. */ +export type RoomHandle

= { + readonly actions: { + [K in keyof RoomActionsOf

]: ( + ...args: ActionArgs[K]> + ) => Promise[K]>>; + }; +}; + +export type RoomOptions

= { + /** Bindings to read instead of the current runtime env, for code outside a request scope. */ + env?: RuntimeEnv; +} & (HasParams

extends true ? { params: RouteParamsOf

} : { params?: never }); + +type RoomActionRequest = { + name: string; + args: Array; + id: string; + params: Record; +}; + +type RoomActionStub = { + __voidRoomAction(request: RoomActionRequest): Promise; +}; + +function resolveRoomBinding( + route: WebSocketRouteMeta, + explicitEnv: RuntimeEnv | undefined, +): DurableObjectNamespace { + let env: RuntimeEnv; + if (explicitEnv) { + env = explicitEnv; + } else { + try { + env = getRawRuntimeEnv(); + } catch { + throw new Error( + 'ws: getRoom() needs a runtime env. Call it from a route, queue, cron or auth hook running inside Void, or pass { env }.', + ); + } + } + const binding = env[route.bindingName] as DurableObjectNamespace | undefined; + if (!binding) { + throw new Error(`ws: Missing Durable Object binding "${route.bindingName}".`); + } + return binding; +} + +function createActionProxy( + stub: RoomActionStub, + id: string, + params: Record, +): Record) => Promise> { + return new Proxy({} as Record) => Promise>, { + get(_target, property) { + // `await` and `Promise.resolve` read `then`. Returning a function there would + // dispatch an action named "then" with resolver callbacks it cannot clone. + if (typeof property !== 'string' || property === 'then') { + return undefined; + } + return (...args: Array) => + stub.__voidRoomAction({ name: property, args, id, params }); + }, + }); +} + +export function getRoom

( + path: HasParams

extends false ? P : never, + options?: RoomOptions

, +): RoomHandle

; +export function getRoom

( + path: P, + options: RoomOptions

, +): RoomHandle

; +export function getRoom( + path: string, + options: { params?: Record; env?: RuntimeEnv } = {}, +): RoomHandle { + const route = __getWebSocketRoute(path); + if (!route) { + throw new Error( + `ws: Unknown WebSocket route "${path}". Use the pattern of a routes/ *.ws.ts file.`, + ); + } + if (route.mode !== 'room') { + throw new Error( + `ws: getRoom() requires a defineRoom() route; "${path}" is a defineWebSocket() route.`, + ); + } + const params = options.params ?? {}; + for (const name of route.params) { + if (typeof params[name] !== 'string') { + throw new TypeError(`ws: Missing route param "${name}" for "${path}".`); + } + } + const binding = resolveRoomBinding(route, options.env); + const id = buildWebSocketInstanceId(route, params); + const stub = binding.get(binding.idFromName(id)) as unknown as RoomActionStub; + return { actions: createActionProxy(stub, id, params) } as RoomHandle; +} diff --git a/packages/void/src/runtime/ws-server.ts b/packages/void/src/runtime/ws-server.ts index 5c0e73551..da84e948b 100644 --- a/packages/void/src/runtime/ws-server.ts +++ b/packages/void/src/runtime/ws-server.ts @@ -1,3 +1,4 @@ +import { DurableObject as CloudflareDurableObject } from 'cloudflare:workers'; import type { StandardSchemaV1 } from '../standard-schema.ts'; import { authContext, type AuthState, type AuthUser } from './auth.ts'; import type { VoidAuthEnv, VoidAuthRuntimeOptions } from './better-auth-shared.ts'; @@ -5,6 +6,7 @@ import { withRuntimeEnv } from './env.ts'; import type { CloudEnv } from './handler.ts'; import type { AnyWebSocketDefinition, + RoomActionContext, RoomContext, RoomController, SocketContext, @@ -19,10 +21,14 @@ export type WebSocketRouteMeta = { params: Array; className: string; bindingName: string; + /** `room` routes accept `getRoom()` actions; `socket` routes do not. */ + mode?: 'room' | 'socket'; }; export type WebSocketDurableObjectInstance = { fetch(request: Request): Promise; + /** Server-only RPC used by `getRoom()`. Never reachable from a client message or HTTP request. */ + __voidRoomAction?(request: RoomActionRequest): Promise; webSocketMessage( webSocket: WebSocket, message: string | ArrayBuffer | ArrayBufferView, @@ -35,6 +41,14 @@ export type WebSocketDurableObjectInstance = { ): Promise; }; +type RoomActionRequest = { + name: string; + args: Array; + /** Instance id the caller addressed, so actions see the same `id` clients do. */ + id: string; + params: Record; +}; + export type WebSocketDurableObjectClass = { new (state: DurableObjectState, env: Record): WebSocketDurableObjectInstance; }; @@ -126,6 +140,48 @@ export function buildWebSocketInstanceId( return route.params.map((name) => `${name}=${params[name] ?? ''}`).join('&'); } +/** Present bracket parameters without changing the router's compiled patterns. */ +function canonicalWebSocketPattern(pattern: string): string { + return pattern + .split('/') + .map((segment) => { + if (segment === '[...]') { + return '*'; + } + const param = segment.match(/^\[\.\.\.([^\]]+)\]$/); + if (param) { + return `:${param[1]}{.+}`; + } + const named = segment.match(/^\[([^\]]+)\]$/); + return named ? `:${named[1]}` : segment; + }) + .join('/'); +} + +/** + * Route metadata for `getRoom()`. The generated route table registers it once per + * worker, so server code can address the same instance a client's `connect()` does. + * Shared through `globalThis` because dev can evaluate this module more than once. + */ +const WEB_SOCKET_ROUTES_SYMBOL: unique symbol = Symbol.for('void.webSocketRoutes.v1'); +const sharedWebSocketRoutes = globalThis as { + [WEB_SOCKET_ROUTES_SYMBOL]?: Map; +}; +const webSocketRoutes = (sharedWebSocketRoutes[WEB_SOCKET_ROUTES_SYMBOL] ??= new Map()); + +/** @internal Called by the generated route table. */ +export function __registerWebSocketRoutes(routes: Array): void { + webSocketRoutes.clear(); + for (const route of routes) { + webSocketRoutes.set(canonicalWebSocketPattern(route.pattern), route); + } +} + +/** @internal Route metadata registered by the generated route table. */ +export function __getWebSocketRoute(pattern: string): WebSocketRouteMeta | undefined { + return webSocketRoutes.get(canonicalWebSocketPattern(pattern)); +} + async function serializeServerEvent(schema: StandardSchemaV1, event: unknown): Promise { const result = await schema['~standard'].validate(event); if (result.issues) { @@ -286,17 +342,44 @@ export function createWebSocketDurableObject( } const instanceRoute = { ...route, key: definition.key }; - class VoidWebSocketDurableObject implements WebSocketDurableObjectInstance { - constructor( - public readonly state: DurableObjectState, - public readonly env: Record, - ) {} + class VoidWebSocketDurableObject + extends CloudflareDurableObject> + implements WebSocketDurableObjectInstance + { + readonly state: DurableObjectState; + + constructor(state: DurableObjectState, env: Record) { + // Extending `DurableObject` enables RPC, which is how `getRoom()` reaches actions. + // Every other method stays `#`-private so it never joins that RPC surface. + super(state, env); + this.state = state; + } fetch(request: Request): Promise { - return withRuntimeEnv(this.env, () => this.handleFetch(request)); + return withRuntimeEnv(this.env, () => this.#handleFetch(request)); + } + + async __voidRoomAction(request: RoomActionRequest): Promise { + return withRuntimeEnv(this.env, () => { + const actions = definition.__kind === 'room' ? definition.actions : undefined; + const action = + actions && Object.hasOwn(actions, request.name) ? actions[request.name] : undefined; + if (typeof action !== 'function') { + throw new Error(`ws: Unknown room action "${request.name}".`); + } + const context: RoomActionContext = { + id: request.id, + params: request.params, + env: this.env as CloudEnv['Bindings'], + storage: this.state.storage, + room: createRoomController(this.state, definition.messages.server), + }; + const handler = action as (context: RoomActionContext, ...args: Array) => unknown; + return handler(context, ...request.args); + }); } - private context( + #context( request: Request, webSocket: WebSocket, stored: StoredConnection, @@ -304,7 +387,7 @@ export function createWebSocketDurableObject( return createHookContext(definition, this.state, this.env, request, webSocket, stored); } - private async handleFetch(request: Request): Promise { + async #handleFetch(request: Request): Promise { const params = getRouteMatch(request, route) ?? {}; const roomId = buildWebSocketInstanceId(instanceRoute, params); const authState = @@ -327,7 +410,7 @@ export function createWebSocketDurableObject( }; setStoredConnection(server, stored); - const beforeConnectCtx = this.context(request, server, stored); + const beforeConnectCtx = this.#context(request, server, stored); const decision = await withAuthState(authState, () => definition.onBeforeConnect?.(beforeConnectCtx as never), ); @@ -337,7 +420,7 @@ export function createWebSocketDurableObject( this.state.acceptWebSocket(server); - const connectCtx = this.context(request, server, stored); + const connectCtx = this.#context(request, server, stored); await withAuthState(authState, () => definition.onConnect?.(connectCtx as never)); return new Response(null, { status: 101, webSocket: client }); @@ -354,7 +437,7 @@ export function createWebSocketDurableObject( user, authState, }; - const ctx = this.context(request, new WebSocketPair()[1], stored); + const ctx = this.#context(request, new WebSocketPair()[1], stored); return withAuthState(authState, () => definition.onRequest!(ctx as never)); } @@ -362,10 +445,10 @@ export function createWebSocketDurableObject( webSocket: WebSocket, message: string | ArrayBuffer | ArrayBufferView, ): Promise { - return withRuntimeEnv(this.env, () => this.handleWebSocketMessage(webSocket, message)); + return withRuntimeEnv(this.env, () => this.#handleWebSocketMessage(webSocket, message)); } - private async handleWebSocketMessage( + async #handleWebSocketMessage( webSocket: WebSocket, message: string | ArrayBuffer | ArrayBufferView, ): Promise { @@ -387,7 +470,7 @@ export function createWebSocketDurableObject( } const request = new Request(`https://void.local${route.pattern}`); - const ctx = this.context(request, webSocket, stored); + const ctx = this.#context(request, webSocket, stored); await withAuthState(stored.authState ?? null, () => definition.onMessage?.(ctx as never, result.value as never), ); @@ -400,11 +483,11 @@ export function createWebSocketDurableObject( wasClean: boolean, ): Promise { return withRuntimeEnv(this.env, () => - this.handleWebSocketClose(webSocket, code, reason, wasClean), + this.#handleWebSocketClose(webSocket, code, reason, wasClean), ); } - private async handleWebSocketClose( + async #handleWebSocketClose( webSocket: WebSocket, code: number, reason: string, @@ -413,7 +496,7 @@ export function createWebSocketDurableObject( completeWebSocketClose(webSocket, code, reason); const stored = getStoredConnection(webSocket); const request = new Request(`https://void.local${route.pattern}`); - const ctx = this.context(request, webSocket, stored); + const ctx = this.#context(request, webSocket, stored); await withAuthState(stored.authState ?? null, () => definition.onClose?.(ctx as never, { code, reason, wasClean }), ); diff --git a/packages/void/src/runtime/ws.ts b/packages/void/src/runtime/ws.ts index 7b49086a6..20b4f5299 100644 --- a/packages/void/src/runtime/ws.ts +++ b/packages/void/src/runtime/ws.ts @@ -57,15 +57,35 @@ export type RoomContext = WebSocketBaseContext & { readonly connection: WebSocketConnection; }; +/** + * Context for a room action. An action runs for other server code, so it has no + * request or user; `id` and `params` describe the addressed room instance. + */ +export type RoomActionContext = { + /** Same instance id a connected client reads as `ctx.id`. */ + readonly id: string; + readonly params: ParamsRecord; + readonly env: CloudEnv['Bindings']; + readonly storage: DurableObjectStorage; + readonly room: RoomController; +}; + +/** + * Named server-side room actions. `Array` accepts every annotated parameter + * list, so each action keeps the parameters and return type its author declares. + * Server events are validated at runtime, but an action's `ctx.room` is not typed + * against `messages.server`. + */ +export type RoomActions = Record< + string, + (context: RoomActionContext, ...args: Array) => unknown +>; + export type SocketContext = WebSocketBaseContext & { readonly socket: WebSocketConnection; }; -export type RoomDefinition< - ClientSchema extends StandardSchemaV1 = StandardSchemaV1, - ServerSchema extends StandardSchemaV1 = StandardSchemaV1, -> = { - readonly __kind: 'room'; +type RoomHooks = { /** Optional persistent resource name. Pin the name reported by `void info` before moving this route. */ readonly name?: string; /** Optional instance key; preserve existing keys when renaming route parameters. */ @@ -91,6 +111,24 @@ export type RoomDefinition< ) => MaybePromise; }; +/** Room definition with actions. Required here so `defineRoom()` can type action parameters. */ +export interface RoomDefinitionInput< + ClientSchema extends StandardSchemaV1, + ServerSchema extends StandardSchemaV1, + Actions extends RoomActions, +> extends RoomHooks { + readonly actions: Actions; +} + +export interface RoomDefinition< + ClientSchema extends StandardSchemaV1 = StandardSchemaV1, + ServerSchema extends StandardSchemaV1 = StandardSchemaV1, + Actions extends RoomActions = {}, +> extends RoomHooks { + readonly __kind: 'room'; + readonly actions?: Actions; +} + export type SocketDefinition< ClientSchema extends StandardSchemaV1 = StandardSchemaV1, ServerSchema extends StandardSchemaV1 = StandardSchemaV1, @@ -121,14 +159,28 @@ export type SocketDefinition< ) => MaybePromise; }; -export type AnyWebSocketDefinition = RoomDefinition | SocketDefinition; +export type AnyWebSocketDefinition = + | RoomDefinition + | SocketDefinition; export function defineRoom< ClientSchema extends StandardSchemaV1, ServerSchema extends StandardSchemaV1, >( - definition: Omit, '__kind'>, -): RoomDefinition { + definition: RoomHooks & { readonly actions?: never }, +): RoomDefinition; +export function defineRoom< + ClientSchema extends StandardSchemaV1, + ServerSchema extends StandardSchemaV1, + const Actions extends RoomActions, +>( + definition: RoomDefinitionInput, +): RoomDefinition; +export function defineRoom( + definition: + | (RoomHooks & { readonly actions?: never }) + | RoomDefinitionInput, +): RoomDefinition { assertResourceName(definition.name); return { ...definition, __kind: 'room' }; } diff --git a/packages/void/test/integration/room-action-types-typecheck.test.ts b/packages/void/test/integration/room-action-types-typecheck.test.ts new file mode 100644 index 000000000..27f0ae0dc --- /dev/null +++ b/packages/void/test/integration/room-action-types-typecheck.test.ts @@ -0,0 +1,182 @@ +import { spawnSync } from 'node:child_process'; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { createRequire } from 'node:module'; +import { tmpdir } from 'node:os'; +import { dirname, join } from 'node:path'; +import { afterEach, describe, expect, it } from 'vite-plus/test'; +import { generateRouteTypes } from '../../src/codegen/route-types.ts'; +import { readProjectPaths } from '../../src/project-paths.ts'; +import type { ScanResult } from '../../src/router/scan.ts'; +import { scanWebSocketRoutesSync } from '../../src/router/websocket.ts'; + +const voidDist = join(import.meta.dirname, '..', '..', 'dist'); +const roots: Array = []; + +afterEach(() => { + for (const root of roots.splice(0)) { + rmSync(root, { recursive: true, force: true }); + } +}); + +const schema = `const schema = { + "~standard": { + version: 1, + vendor: "test", + validate: (value: unknown) => ({ value }), + }, +} as const;`; + +const relayRoute = `import { defineRoom } from 'void/ws'; +${schema} +export default defineRoom({ + messages: { client: schema, server: schema }, + actions: { + revokeSession(context, sessionId: string, code: number) { + return { closed: 1, sessionId, code, id: context.id }; + }, + ping(context) { + return context.id; + }, + }, +}); +`; + +const plainRoute = `import { defineRoom } from 'void/ws'; +${schema} +export default defineRoom({ messages: { client: schema, server: schema } }); +`; + +const feedRoute = `import { defineWebSocket } from 'void/ws'; +${schema} +export default defineWebSocket({ messages: { client: schema, server: schema } }); +`; + +const check = `import { getRoom, type RoomHandle } from 'void/ws/server'; + +const room = getRoom('/relay/[accountId]', { params: { accountId: 'acct_1' } }); +const handle: RoomHandle<'/relay/[accountId]'> = room; +const revoked: Promise<{ closed: number; sessionId: string; code: number; id: string }> = + room.actions.revokeSession('session_1', 4001); +const pinged: Promise = room.actions.ping(); + +const plain = getRoom('/plain'); +const noActions: {} = plain.actions; + +// @ts-expect-error routes with parameters require them +getRoom('/relay/[accountId]'); +// @ts-expect-error parameter names come from the route +getRoom('/relay/[accountId]', { params: { accountId: 'acct_1', extra: 'x' } }); +// @ts-expect-error action arguments are typed +room.actions.revokeSession(1, 4001); +// @ts-expect-error action arguments are required +room.actions.revokeSession('session_1'); +// @ts-expect-error only declared actions exist +room.actions.missing(); +// @ts-expect-error defineWebSocket() routes expose no actions +getRoom('/feed'); +// @ts-expect-error a room without actions has none +plain.actions.missing(); +`; + +/** + * Typecheck `getRoom()` against the declarations `generateRouteTypes()` emits for + * real route files, so codegen and the public handle types cannot drift apart + * without failing here. + */ +function typecheck( + routes: Record, + files: Record, + transformDts?: (dts: string) => string, +) { + const root = mkdtempSync(join(tmpdir(), 'void-room-action-types-')); + roots.push(root); + const routesDir = join(root, 'routes'); + for (const [path, content] of Object.entries(routes)) { + mkdirSync(dirname(join(routesDir, path)), { recursive: true }); + writeFileSync(join(routesDir, path), content); + } + + const websockets = scanWebSocketRoutesSync(root, readProjectPaths(root)); + const { dts } = generateRouteTypes( + { + routes: [], + websockets, + middleware: [], + jobs: [], + queues: [], + pages: { pages: [], layouts: [], namedLayouts: [] }, + } satisfies ScanResult, + routesDir, + join(root, '.void'), + ); + + const all: Record = { + 'tsconfig.json': JSON.stringify({ + compilerOptions: { + target: 'ES2022', + module: 'NodeNext', + moduleResolution: 'NodeNext', + lib: ['ES2022'], + types: [], + strict: true, + noEmit: true, + skipLibCheck: true, + allowImportingTsExtensions: true, + paths: { + void: [join(voidDist, 'index.d.mts')], + 'void/ws': [join(voidDist, 'runtime/ws.d.mts')], + 'void/ws/server': [join(voidDist, 'runtime/ws-room.d.mts')], + // Generated declarations augment this base module; the package helper + // aliases must not hide defects in the generated output. + 'void/routes': [join(root, 'routes-base.d.ts')], + }, + }, + files: ['.void/routes.d.ts', 'check.ts'], + }), + '.void/routes.d.ts': transformDts ? transformDts(dts) : dts, + 'routes-base.d.ts': 'export {};', + ...files, + }; + for (const [path, content] of Object.entries(all)) { + mkdirSync(dirname(join(root, path)), { recursive: true }); + writeFileSync(join(root, path), content); + } + + const tsc = join( + dirname(createRequire(import.meta.url).resolve('typescript/package.json')), + 'bin/tsc', + ); + return spawnSync(process.execPath, [tsc, '-p', join(root, 'tsconfig.json')], { + encoding: 'utf8', + }); +} + +describe('getRoom() against generated route types', () => { + it('accepts the actions the generated declarations expose', () => { + const result = typecheck( + { + 'relay/[accountId].ws.ts': relayRoute, + 'plain.ws.ts': plainRoute, + 'feed.ws.ts': feedRoute, + }, + { 'check.ts': check }, + ); + expect(result.stdout + result.stderr).toBe(''); + expect(result.status).toBe(0); + }); + + it('catches a declaration that drops the generated actions', () => { + const result = typecheck( + { + 'relay/[accountId].ws.ts': relayRoute, + 'plain.ws.ts': plainRoute, + 'feed.ws.ts': feedRoute, + }, + { 'check.ts': check }, + // Same check, but the generated map no longer exposes the room's actions. + // The assertions above must fail, so this proves they are live. + (dts) => dts.replace(/^\s*actions: .*$/m, ''), + ); + expect(result.status).not.toBe(0); + }); +}); diff --git a/packages/void/test/integration/websocket-room-actions.test.ts b/packages/void/test/integration/websocket-room-actions.test.ts new file mode 100644 index 000000000..87048f13e --- /dev/null +++ b/packages/void/test/integration/websocket-room-actions.test.ts @@ -0,0 +1,453 @@ +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { Miniflare, convertV4MiniflareOptions } from 'miniflare'; +import { build } from 'vite'; +import { afterAll, beforeAll, describe, expect, it } from 'vite-plus/test'; +import { compileRouteTable, compileWorkerEntry } from '../../src/router/compile.ts'; +import type { ScanResult } from '../../src/router/scan.ts'; + +const root = join(import.meta.dirname, '..', '..', '..', '..'); +const temporary = mkdtempSync(join(tmpdir(), 'void-room-actions-')); +const output = join(temporary, 'dist'); +const routeTablePath = join(temporary, 'virtual-void-routes.ts'); +const workerPath = join(temporary, 'worker.ts'); +const runtimeEnvPath = join(root, 'packages/void/src/runtime/env.ts'); +const roomRuntimePath = join(root, 'packages/void/src/runtime/ws-room.ts'); + +type MiniflareInstance = { + ready: Promise; + dispatchFetch(input: string, init?: RequestInit): Promise; + getQueueProducer( + bindingName: string, + ): Promise<{ send(body: Body): Promise }>; + dispose(): Promise; +}; + +let miniflare: MiniflareInstance; + +async function connect(url: URL, accountId: string, session: string) { + const response = await miniflare.dispatchFetch( + new URL(`/relay/${accountId}?session=${session}`, url).toString(), + { headers: { Upgrade: 'websocket' } }, + ); + expect(response.status).toBe(101); + const socket = response.webSocket!; + socket.accept(); + const closed = new Promise((resolve) => { + socket.addEventListener('close', (event) => resolve(event), { once: true }); + }); + return { socket, closed }; +} + +async function call(url: URL, path: string, params: Record) { + const target = new URL(path, url); + for (const [key, value] of Object.entries(params)) { + target.searchParams.set(key, value); + } + return miniflare.dispatchFetch(target.toString()); +} + +function sleep(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function waitFor(predicate: () => boolean, timeoutMs = 2_000) { + const started = Date.now(); + while (!predicate()) { + if (Date.now() - started > timeoutMs) { + return false; + } + await sleep(10); + } + return true; +} + +describe('server-side room actions through the generated Worker', () => { + beforeAll(async () => { + const routesDir = join(temporary, 'routes'); + const middlewareDir = join(temporary, 'middleware'); + mkdirSync(routesDir, { recursive: true }); + mkdirSync(middlewareDir, { recursive: true }); + + writeFileSync( + join(routesDir, 'relay.ws.ts'), + `const schema = { "~standard": { version: 1, vendor: "test", validate: (value) => ({ value }) } }; +export default { + __kind: "room", + messages: { client: schema, server: schema }, + onConnect(ctx) { + ctx.connection.setState({ sessionId: new URL(ctx.request.url).searchParams.get("session") }); + }, + async onMessage(ctx) { + const seen = ((await ctx.storage.get("messages")) ?? 0) + 1; + await ctx.storage.put("messages", seen); + }, + actions: { + async revokeSession(ctx, sessionId, code) { + let closed = 0; + for (const connection of ctx.room.getConnections()) { + if (connection.state?.sessionId === sessionId) { + connection.close(code, "session revoked"); + closed += 1; + } + } + const calls = ((await ctx.storage.get("revocations")) ?? 0) + 1; + await ctx.storage.put("revocations", calls); + return { closed, calls, id: ctx.id, params: ctx.params }; + }, + async broadcast(ctx, text) { + await ctx.room.broadcast({ type: "relay.message", text }); + return { sent: true }; + }, + async stats(ctx) { + return { + messages: (await ctx.storage.get("messages")) ?? 0, + hasBinding: typeof ctx.env.RELAY_WS === "object", + }; + }, + }, +}; +`, + ); + + const handler = (body: string) => + `import { getRoom } from ${JSON.stringify(roomRuntimePath)}; +export async function GET(c) { + const url = new URL(c.req.url); + const accountId = url.searchParams.get("accountId"); + try { + const room = getRoom("/relay/:accountId", { params: { accountId } }); + return Response.json(await room.actions.${body}); + } catch (error) { + return Response.json({ error: error.message }, { status: 500 }); + } +} +`; + writeFileSync( + join(routesDir, 'revoke.ts'), + handler( + 'revokeSession(url.searchParams.get("session"), Number(url.searchParams.get("code") ?? 4001))', + ), + ); + writeFileSync( + join(routesDir, 'broadcast.ts'), + handler('broadcast(url.searchParams.get("text"))'), + ); + writeFileSync(join(routesDir, 'stats.ts'), handler('stats()')); + writeFileSync(join(routesDir, 'unknown.ts'), handler('missingAction()')); + + const roomImport = `import { getRoom } from ${JSON.stringify(roomRuntimePath)};`; + writeFileSync( + join(routesDir, 'assimilate.ts'), + `${roomImport} +export async function GET(c) { + const accountId = new URL(c.req.url).searchParams.get("accountId"); + const room = getRoom("/relay/:accountId", { params: { accountId } }); + const awaited = await room.actions; + const resolved = await Promise.resolve(room.actions); + return Response.json({ awaited: awaited === room.actions, resolved: resolved === room.actions }); +} +`, + ); + writeFileSync( + join(routesDir, 'helpers.ts'), + `export async function GET(c) { + const namespace = c.env.RELAY_WS; + const stub = namespace.get(namespace.idFromName("accountId=acct_1")); + const errors = {}; + for (const name of ["context", "handleFetch", "handleWebSocketMessage", "handleWebSocketClose"]) { + try { + await stub[name](); + errors[name] = "callable"; + } catch (error) { + errors[name] = error.message; + } + } + const stats = await stub.__voidRoomAction({ + name: "stats", + args: [], + id: "accountId=acct_1", + params: { accountId: "acct_1" }, + }); + return Response.json({ errors, stats }); +} +`, + ); + writeFileSync( + join(temporary, 'cron.ts'), + `${roomImport} +export const cron = "*/5 * * * *"; +export default async function () { + const room = getRoom("/relay/:accountId", { params: { accountId: "acct_8" } }); + await room.actions.broadcast("from cron"); +} +`, + ); + writeFileSync( + join(temporary, 'queue.ts'), + `${roomImport} +export default async function (batch) { + const body = batch.messages[0].body; + const room = getRoom("/relay/:accountId", { params: { accountId: body.accountId } }); + await room.actions.broadcast(body.text); +} +`, + ); + + const routeFile = (pattern: string, filePath: string) => ({ + pattern, + methods: ['GET'], + params: [], + catchAll: false, + env: null, + filePath, + }); + const scan: ScanResult = { + routes: [ + routeFile('/revoke', 'revoke.ts'), + routeFile('/broadcast', 'broadcast.ts'), + routeFile('/stats', 'stats.ts'), + routeFile('/unknown', 'unknown.ts'), + routeFile('/assimilate', 'assimilate.ts'), + routeFile('/helpers', 'helpers.ts'), + ], + websockets: [ + { + pattern: '/relay/:accountId', + methods: [], + params: ['accountId'], + catchAll: false, + env: null, + filePath: 'relay.ws.ts', + className: 'RelayWs', + bindingName: 'RELAY_WS', + mode: 'room', + hasAuthGate: false, + }, + ], + middleware: [], + jobs: [{ filePath: 'cron.ts', name: 'cron', crons: ['*/5 * * * *'] }], + queues: [{ filePath: 'queue.ts', name: 'relay-jobs', bindingName: 'QUEUE_RELAY_JOBS' }], + pages: { pages: [], layouts: [], namedLayouts: [] }, + }; + + writeFileSync( + routeTablePath, + compileRouteTable(scan, routesDir, middlewareDir, temporary, temporary, null, { + honoPath: import.meta.resolve('hono'), + honoRoutePath: import.meta.resolve('hono/route'), + runtimeEnvPath, + webSocketRuntimePath: join(root, 'packages/void/src/runtime/ws-server.ts'), + projectRoot: temporary, + homeDir: temporary, + reactRefresh: false, + auth: null, + }), + ); + writeFileSync( + workerPath, + compileWorkerEntry({ webSocketClasses: ['RelayWs'], runtimeEnvPath, hasQueues: true }), + ); + + await build({ + configFile: false, + logLevel: 'silent', + plugins: [ + { + name: 'void-room-actions-test', + resolveId: (id) => (id === 'virtual:void-routes' ? routeTablePath : null), + }, + ], + ssr: { noExternal: true }, + build: { + emptyOutDir: true, + outDir: output, + ssr: workerPath, + target: 'esnext', + rollupOptions: { + external: [/^cloudflare:/, /^node:/], + output: { entryFileNames: 'worker.mjs', codeSplitting: false }, + }, + }, + }); + + miniflare = new Miniflare( + convertV4MiniflareOptions({ + compatibilityDate: '2026-07-01', + compatibilityFlags: ['nodejs_compat'], + modules: true, + modulesRoot: output, + scriptPath: join(output, 'worker.mjs'), + bindings: { __VOID_PROXY_TOKEN: 'test-token' }, + durableObjects: { RELAY_WS: { className: 'RelayWs', useSQLite: true } }, + queueProducers: { JOBS: 'relay-jobs' }, + queueConsumers: { 'relay-jobs': { maxBatchTimeout: 0.05 } }, + }), + ) as unknown as MiniflareInstance; + }, 30_000); + + afterAll(async () => { + await miniflare?.dispose(); + rmSync(temporary, { recursive: true, force: true }); + }); + + it('reports the same instance id and params a connected client sees', async () => { + const url = await miniflare.ready; + const first = await call(url, '/stats', { accountId: 'acct_1' }); + expect(first.status).toBe(200); + + const other = await call(url, '/broadcast', { accountId: 'acct_2', text: 'nobody' }); + expect(await other.json()).toEqual({ sent: true }); + + const response = await call(url, '/revoke', { accountId: 'acct_1', session: 'missing' }); + expect(await response.json()).toEqual({ + closed: 0, + calls: 1, + id: 'accountId=acct_1', + params: { accountId: 'acct_1' }, + }); + }); + + it('closes only the matching connections and keeps room storage', async () => { + const url = await miniflare.ready; + const target = await connect(url, 'acct_1', 'session_1'); + const bystander = await connect(url, 'acct_1', 'session_2'); + const elsewhere = await connect(url, 'acct_3', 'session_1'); + + const response = await call(url, '/revoke', { + accountId: 'acct_1', + session: 'session_1', + code: '4001', + }); + expect(await response.json()).toMatchObject({ closed: 1, calls: 2 }); + + const closeEvent = await target.closed; + expect(closeEvent.code).toBe(4001); + expect(closeEvent.reason).toBe('session revoked'); + expect((await Promise.race([bystander.closed, sleep(50).then(() => 'open')])) as unknown).toBe( + 'open', + ); + expect((await Promise.race([elsewhere.closed, sleep(10).then(() => 'open')])) as unknown).toBe( + 'open', + ); + + const again = await call(url, '/revoke', { accountId: 'acct_1', session: 'session_1' }); + expect(await again.json()).toMatchObject({ closed: 0, calls: 3 }); + + bystander.socket.close(1000, 'done'); + elsewhere.socket.close(1000, 'done'); + await Promise.all([bystander.closed, elsewhere.closed]); + }); + + it('broadcasts from an action to the addressed room only', async () => { + const url = await miniflare.ready; + const first = await connect(url, 'acct_4', 'session_1'); + const second = await connect(url, 'acct_4', 'session_2'); + const other = await connect(url, 'acct_5', 'session_1'); + + const messages: Array = []; + first.socket.addEventListener('message', (event) => messages.push(String(event.data))); + second.socket.addEventListener('message', (event) => messages.push(String(event.data))); + other.socket.addEventListener('message', (event) => messages.push(String(event.data))); + + const response = await call(url, '/broadcast', { accountId: 'acct_4', text: 'hello' }); + expect(await response.json()).toEqual({ sent: true }); + + expect(await waitFor(() => messages.length === 2)).toBe(true); + expect(messages).toEqual([ + JSON.stringify({ type: 'relay.message', text: 'hello' }), + JSON.stringify({ type: 'relay.message', text: 'hello' }), + ]); + + first.socket.close(1000, 'done'); + second.socket.close(1000, 'done'); + other.socket.close(1000, 'done'); + await Promise.all([first.closed, second.closed, other.closed]); + }); + + it('keeps internal helpers off the RPC surface while intended RPC still works', async () => { + const url = await miniflare.ready; + const response = await call(url, '/helpers', {}); + expect(response.status).toBe(200); + const body = (await response.json()) as { + errors: Record; + stats: { messages: number; hasBinding: boolean }; + }; + for (const name of [ + 'context', + 'handleFetch', + 'handleWebSocketMessage', + 'handleWebSocketClose', + ]) { + expect(body.errors[name]).toMatch(/does not implement the method/i); + } + expect(body.stats.hasBinding).toBe(true); + expect(typeof body.stats.messages).toBe('number'); + }); + + it('stays out of Promise assimilation inside the worker', async () => { + const url = await miniflare.ready; + const response = await call(url, '/assimilate', { accountId: 'acct_1' }); + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ awaited: true, resolved: true }); + }); + + it('runs actions from cron and queue handlers', async () => { + const url = await miniflare.ready; + const target = await connect(url, 'acct_8', 'session_1'); + const messages: Array = []; + target.socket.addEventListener('message', (event) => messages.push(String(event.data))); + + const scheduled = await miniflare.dispatchFetch(new URL('/__void/scheduled', url).toString(), { + method: 'POST', + headers: { 'x-void-internal': 'test-token' }, + body: JSON.stringify({ cron: '*/5 * * * *', scheduledTime: Date.now() }), + }); + expect(scheduled.status).toBe(200); + expect(await waitFor(() => messages.length === 1)).toBe(true); + + const producer = await miniflare.getQueueProducer<{ accountId: string; text: string }>('JOBS'); + await producer.send({ accountId: 'acct_8', text: 'from queue' }); + expect(await waitFor(() => messages.length === 2)).toBe(true); + + expect(messages).toEqual([ + JSON.stringify({ type: 'relay.message', text: 'from cron' }), + JSON.stringify({ type: 'relay.message', text: 'from queue' }), + ]); + + target.socket.close(1000, 'done'); + await target.closed; + }); + + it('reports unknown actions without running anything', async () => { + const url = await miniflare.ready; + const response = await call(url, '/unknown', { accountId: 'acct_6' }); + expect(response.status).toBe(500); + expect(await response.json()).toEqual({ error: 'ws: Unknown room action "missingAction".' }); + }); + + it('cannot be reached from client messages or ordinary HTTP', async () => { + const url = await miniflare.ready; + const target = await connect(url, 'acct_7', 'session_1'); + + target.socket.send( + JSON.stringify({ name: 'revokeSession', args: ['session_1', 4001], id: 'accountId=acct_7' }), + ); + expect((await Promise.race([target.closed, sleep(100).then(() => 'open')])) as unknown).toBe( + 'open', + ); + const stats = await call(url, '/stats', { accountId: 'acct_7' }); + expect(await stats.json()).toEqual({ messages: 1, hasBinding: true }); + + const http = await miniflare.dispatchFetch(new URL('/relay/acct_7', url).toString(), { + method: 'POST', + headers: { 'x-void-room-action': 'revokeSession' }, + body: JSON.stringify({ sessionId: 'session_1' }), + }); + expect(http.status).toBe(426); + + target.socket.close(1000, 'done'); + await target.closed; + }); +}); diff --git a/packages/void/test/unit/compile.test.ts b/packages/void/test/unit/compile.test.ts index 219c7f521..51529c46b 100644 --- a/packages/void/test/unit/compile.test.ts +++ b/packages/void/test/unit/compile.test.ts @@ -167,6 +167,11 @@ describe('compileRouteTable', () => { expect(code).toContain('createWebSocketDurableObject'); expect(code).toContain('export const ChatRoomWs = __createWebSocketDurableObject'); expect(code).toContain('const __voidWebSocketRoutes = [{"pattern":"/chat/:room"'); + expect(code).toContain('"mode":"room"'); + expect(code).toContain('__registerWebSocketRoutes(__voidWebSocketRoutes);'); + expect(code.indexOf('__voidWebSocketRoutes[0].key = websocket0.key')).toBeLessThan( + code.indexOf('__registerWebSocketRoutes(__voidWebSocketRoutes);'), + ); expect(code).toContain('app.all("/chat/:room"'); expect(code).toContain( 'if (c.req.raw.headers.get("upgrade")?.toLowerCase() !== "websocket") return next()', diff --git a/packages/void/test/unit/resource-name-runtime.test.ts b/packages/void/test/unit/resource-name-runtime.test.ts index ea6d6597c..f1004cfea 100644 --- a/packages/void/test/unit/resource-name-runtime.test.ts +++ b/packages/void/test/unit/resource-name-runtime.test.ts @@ -32,9 +32,11 @@ describe('persistent resource runtime contracts', () => { ).not.toThrow(); }); - it.each([defineRoom, defineWebSocket])( - 'requires generated WebSocket metadata to match the authored name', - (define) => { + it.each(['room', 'socket'] as const)( + 'requires generated WebSocket metadata to match the authored %s name', + (kind) => { + const define = (definition: { name?: string; messages: typeof messages }) => + kind === 'room' ? defineRoom(definition) : defineWebSocket(definition); const definition = define({ name: 'chat-room', messages }); expect(() => createWebSocketDurableObject(definition, route)).toThrow( 'statically discoverable', diff --git a/packages/void/test/unit/route-types.test.ts b/packages/void/test/unit/route-types.test.ts index 42d4002a4..7e8fe662d 100644 --- a/packages/void/test/unit/route-types.test.ts +++ b/packages/void/test/unit/route-types.test.ts @@ -447,6 +447,34 @@ describe('generateRouteTypes', () => { expect(dts).toContain('client:'); expect(dts).toContain('server:'); expect(dts).toContain('params: RouteParams<["room"]>'); + expect(dts).toContain( + 'actions: NonNullable<(typeof import("../routes/chat/[room].ws.ts"))["default"]["actions"]>;', + ); + }); + + it('omits actions for socket routes', () => { + const scan = makeScan({ + websockets: [ + { + pattern: '/feed', + methods: [], + params: [], + catchAll: false, + env: null, + filePath: 'feed.ws.ts', + className: 'FeedWs', + bindingName: 'FeedWs', + mode: 'socket', + hasAuthGate: false, + }, + ], + }); + + const { dts } = generateRouteTypes(scan, '/project/routes', '/project/.void'); + const websocketMap = dts.split(' interface WebSocketRouteMap {')[1]!.split('\n }\n}')[0]!; + + expect(websocketMap).toContain('params: RouteParams<[]>'); + expect(websocketMap).not.toContain('actions:'); }); it('emits one WebSocket route key for colliding route files', () => { @@ -484,6 +512,7 @@ describe('generateRouteTypes', () => { expect(websocketMap).toContain('chat/[room].prod.ws.ts'); expect(websocketMap).toMatch(/client: .*dev\.ws\.ts.* & .*prod\.ws\.ts/); expect(websocketMap).toMatch(/server: .*dev\.ws\.ts.* \| .*prod\.ws\.ts/); + expect(websocketMap).toMatch(/actions: .*dev\.ws\.ts.* \| .*prod\.ws\.ts/); }); it('skips pages without action in PageActionMap', () => { diff --git a/packages/void/test/unit/ws-room-actions.test.ts b/packages/void/test/unit/ws-room-actions.test.ts new file mode 100644 index 000000000..88de9a3f7 --- /dev/null +++ b/packages/void/test/unit/ws-room-actions.test.ts @@ -0,0 +1,180 @@ +import { describe, expect, it, vi } from 'vite-plus/test'; +import { getRoom } from '../../src/runtime/ws-room.ts'; +import { + createWebSocketDurableObject, + forwardWebSocketRequest, + __registerWebSocketRoutes, +} from '../../src/runtime/ws-server.ts'; +import { defineRoom, defineWebSocket } from '../../src/runtime/ws.ts'; +import type { StandardSchemaV1 } from '../../src/standard-schema.ts'; + +vi.mock('cloudflare:workers', () => ({ DurableObject: class {} })); + +const schema: StandardSchemaV1 = { + '~standard': { version: 1, vendor: 'test', validate: (value) => ({ value }) }, +}; +const messages = { client: schema, server: schema }; +const route = { + pattern: '/relay/:accountId', + params: ['accountId'], + className: 'RelayWs', + bindingName: 'RELAY_WS', + mode: 'room' as const, +}; + +const getRoomLoose = getRoom as unknown as ( + path: string, + options?: { params?: Record; env?: Record }, +) => { actions: Record) => Promise> }; + +function createEnv( + handler: (request: { + name: string; + args: Array; + id: string; + params: Record; + }) => Promise, +) { + const keys: Array = []; + const env = { + RELAY_WS: { + idFromName(name: string) { + keys.push(name); + return name; + }, + get: () => ({ __voidRoomAction: handler }), + }, + }; + return { env, keys }; +} + +describe('server-side room actions', () => { + it('accepts bracket patterns and custom instance keys', async () => { + const key = ({ params }: { params: Record }) => `account=${params.id}`; + const keyedRoute = { ...route, pattern: '/rooms/:id', params: ['id'], key }; + __registerWebSocketRoutes([keyedRoute]); + const { env, keys } = createEnv(async () => null); + + await getRoomLoose('/rooms/[id]', { params: { id: 'acct_1' }, env }).actions.revokeSession('s'); + + const clientKeys: Array = []; + await forwardWebSocketRequest( + new Request('https://app.test/rooms/acct_1'), + { + RELAY_WS: { + idFromName(name: string) { + clientKeys.push(name); + return name; + }, + get: () => ({ fetch: async () => new Response('ok') }), + }, + }, + keyedRoute, + ); + + expect(keys).toEqual(['account=acct_1']); + expect(clientKeys).toEqual(keys); + }); + + it('addresses catch-all patterns by their public spelling', async () => { + const named = { ...route, pattern: '/files/:path{.+}', params: ['path'] }; + __registerWebSocketRoutes([named]); + const { env, keys } = createEnv(async () => null); + + await getRoomLoose('/files/[...path]', { params: { path: 'a/b' }, env }).actions.sync(); + expect(keys).toEqual(['path=a/b']); + + __registerWebSocketRoutes([{ ...route, pattern: '/*', params: [] }]); + await getRoomLoose('/[...]', { env }).actions.sync(); + expect(keys).toEqual(['path=a/b', 'default']); + }); + + it('is not thenable, so awaiting it never dispatches an action named "then"', async () => { + __registerWebSocketRoutes([route]); + const calls: Array = []; + const { env } = createEnv(async (request) => { + calls.push(request.name); + throw new Error(`unexpected action "${request.name}"`); + }); + const room = getRoomLoose('/relay/:accountId', { params: { accountId: 'acct_1' }, env }); + + await expect(Promise.resolve(room.actions)).resolves.toBe(room.actions); + await expect((async () => room.actions)()).resolves.toBe(room.actions); + expect(calls).toEqual([]); + }); + + it('rejects unknown routes, socket routes and missing params', () => { + __registerWebSocketRoutes([ + route, + { + pattern: '/feed', + params: [], + className: 'FeedWs', + bindingName: 'FEED_WS', + mode: 'socket', + }, + ]); + const { env } = createEnv(async () => null); + + expect(() => getRoomLoose('/missing', { env })).toThrow('Unknown WebSocket route'); + expect(() => getRoomLoose('/feed', { env })).toThrow('defineRoom() route'); + expect(() => getRoomLoose('/relay/:accountId', { params: {}, env })).toThrow( + 'Missing route param "accountId"', + ); + }); + + it('needs a runtime env or an explicit one', () => { + __registerWebSocketRoutes([route]); + expect(() => getRoomLoose('/relay/:accountId', { params: { accountId: 'a' } })).toThrow( + 'getRoom() needs a runtime env', + ); + expect(() => + getRoomLoose('/relay/:accountId', { params: { accountId: 'a' }, env: {} }), + ).toThrow('Missing Durable Object binding "RELAY_WS"'); + }); +}); + +describe('room action dispatch inside the Durable Object', () => { + const state = {} as unknown as DurableObjectState; + const definition = defineRoom({ + messages, + actions: { + fail() { + throw new Error('deliberate failure'); + }, + }, + }); + const DurableObject = createWebSocketDurableObject(definition, route); + const instance = new DurableObject(state, {}); + + it('reports unknown actions and handler errors', async () => { + await expect( + instance.__voidRoomAction!({ name: 'missing', args: [], id: 'id', params: {} }), + ).rejects.toThrow('Unknown room action "missing"'); + await expect( + instance.__voidRoomAction!({ name: 'constructor', args: [], id: 'id', params: {} }), + ).rejects.toThrow('Unknown room action "constructor"'); + await expect( + instance.__voidRoomAction!({ name: 'fail', args: [], id: 'id', params: {} }), + ).rejects.toThrow('deliberate failure'); + }); + + it('keeps socket routes out of actions', async () => { + const SocketObject = createWebSocketDurableObject(defineWebSocket({ messages }), { + ...route, + pattern: '/feed', + params: [], + className: 'FeedWs', + bindingName: 'FEED_WS', + }); + + await expect( + new SocketObject(state, {}).__voidRoomAction!({ + name: 'revokeSession', + args: ['s'], + id: 'default', + params: {}, + }), + ).rejects.toThrow('Unknown room action "revokeSession"'); + }); +}); diff --git a/packages/void/test/unit/ws-room.typecheck.ts b/packages/void/test/unit/ws-room.typecheck.ts new file mode 100644 index 000000000..dddb1dc7f --- /dev/null +++ b/packages/void/test/unit/ws-room.typecheck.ts @@ -0,0 +1,80 @@ +import { expectTypeOf } from 'vite-plus/test'; +import { getRoom, type RoomHandle } from '../../src/runtime/ws-room.ts'; +import { defineRoom, type RoomActionContext } from '../../src/runtime/ws.ts'; +import type { StandardSchemaV1 } from '../../src/standard-schema.ts'; + +declare module 'void/routes' { + interface WebSocketRouteMap { + '/relay/[accountId]': { + params: { accountId: string }; + client: { type: 'relay.ping' }; + server: { type: 'relay.closed'; closed: number }; + actions: { + revokeSession: ( + context: RoomActionContext, + sessionId: string, + code: number, + ) => { closed: number }; + ping: (context: RoomActionContext) => string; + }; + }; + '/feed': { + params: Record; + client: { type: 'feed.ack' }; + server: { type: 'feed.item' }; + }; + } +} + +declare const client: StandardSchemaV1<{ type: 'relay.ping' }>; +declare const server: StandardSchemaV1<{ type: 'relay.closed'; closed: number }>; + +const definition = defineRoom({ + messages: { client, server }, + actions: { + revokeSession(context, sessionId: string, code: number) { + expectTypeOf(context.id).toEqualTypeOf(); + expectTypeOf(context.params).toEqualTypeOf>(); + expectTypeOf(context.room.getConnections).toBeFunction(); + for (const connection of context.room.getConnections()) { + connection.close(code, 'revoked'); + } + return { closed: 1, sessionId }; + }, + ping(context) { + return context.id; + }, + }, +}); + +expectTypeOf(definition.actions!.revokeSession).toEqualTypeOf< + ( + context: RoomActionContext, + sessionId: string, + code: number, + ) => { closed: number; sessionId: string } +>(); +expectTypeOf(definition.actions!.ping).toEqualTypeOf<(context: RoomActionContext) => string>(); + +const plain = defineRoom({ messages: { client, server } }); +expectTypeOf(plain.actions).toEqualTypeOf<{} | undefined>(); + +const room = getRoom('/relay/[accountId]', { params: { accountId: 'acct_1' } }); +expectTypeOf(room).toEqualTypeOf>(); +expectTypeOf(room.actions.revokeSession).toEqualTypeOf< + (sessionId: string, code: number) => Promise<{ closed: number }> +>(); +expectTypeOf(room.actions.ping).toEqualTypeOf<() => Promise>(); + +// @ts-expect-error Routes with parameters require them. +getRoom('/relay/[accountId]'); +// @ts-expect-error Parameter names come from the route. +getRoom('/relay/[accountId]', { params: { accountId: 'acct_1', extra: 'x' } }); +// @ts-expect-error Action arguments are typed. +room.actions.revokeSession(1, 4001); +// @ts-expect-error Action arguments are required. +room.actions.revokeSession('session_1'); +// @ts-expect-error Only declared actions exist. +room.actions.missing(); +// @ts-expect-error defineWebSocket() routes expose no actions. +getRoom('/feed'); diff --git a/packages/void/test/unit/ws-server-close.test.ts b/packages/void/test/unit/ws-server-close.test.ts index e1a10b0fc..bfb0e53b0 100644 --- a/packages/void/test/unit/ws-server-close.test.ts +++ b/packages/void/test/unit/ws-server-close.test.ts @@ -3,9 +3,11 @@ import { createWebSocketDurableObject, createWebSocketTombstoneDurableObject, } from '../../src/runtime/ws-server.ts'; -import { defineRoom, defineWebSocket } from '../../src/runtime/ws.ts'; +import { defineRoom, defineWebSocket, type WebSocketBaseContext } from '../../src/runtime/ws.ts'; import type { StandardSchemaV1 } from '../../src/standard-schema.ts'; +vi.mock('cloudflare:workers', () => ({ DurableObject: class {} })); + const schema: StandardSchemaV1 = { '~standard': { version: 1, vendor: 'test', validate: (value) => ({ value }) }, }; @@ -27,10 +29,22 @@ function socket(readyState = 2) { return { webSocket, close }; } +function define( + kind: 'room' | 'socket', + definition: { + messages: { client: StandardSchemaV1; server: StandardSchemaV1 }; + onClose: ( + context: WebSocketBaseContext, + details?: { code: number; reason: string; wasClean: boolean }, + ) => void; + }, +) { + return kind === 'room' ? defineRoom(definition) : defineWebSocket(definition); +} + it.each(['room', 'socket'] as const)( 'completes the close handshake before invoking %s hooks', async (kind) => { - const define = kind === 'room' ? defineRoom : defineWebSocket; const { webSocket, close } = socket(); const onClose = vi.fn( (_context: unknown, details?: { code: number; reason: string; wasClean: boolean }) => { @@ -39,7 +53,7 @@ it.each(['room', 'socket'] as const)( }, ); const DurableObject = createWebSocketDurableObject( - define({ messages: { client: schema, server: schema }, onClose }), + define(kind, { messages: { client: schema, server: schema }, onClose }), route, ); await new DurableObject(state, {}).webSocketClose(webSocket, 1000, 'done', true); diff --git a/packages/void/vite.config.ts b/packages/void/vite.config.ts index 71c0c6b37..e44de785b 100644 --- a/packages/void/vite.config.ts +++ b/packages/void/vite.config.ts @@ -87,6 +87,7 @@ export default defineConfig({ 'src/runtime/auth.ts', 'src/runtime/auth-client.ts', 'src/runtime/ws.ts', + 'src/runtime/ws-room.ts', 'src/runtime/auth-client-react.ts', 'src/runtime/auth-client-vue.ts', 'src/runtime/auth-client-svelte.ts',