diff --git a/docs/.vitepress/config.ts b/docs/.vitepress/config.ts index 6f7695335..ad2f1113e 100644 --- a/docs/.vitepress/config.ts +++ b/docs/.vitepress/config.ts @@ -226,6 +226,7 @@ export default extendConfig( { text: 'Authentication', link: '/guide/auth' }, { text: 'WebSockets', link: '/guide/websockets' }, { text: 'Server-Sent Events', link: '/guide/sse' }, + { text: 'Live Event Streams', link: '/guide/live' }, { text: 'Cron Jobs', link: '/guide/jobs' }, { text: 'Queues', link: '/guide/queues' }, { text: 'Remote Development', link: '/guide/remote-dev' }, diff --git a/docs/guide/live.md b/docs/guide/live.md new file mode 100644 index 000000000..cce52c672 --- /dev/null +++ b/docs/guide/live.md @@ -0,0 +1,230 @@ +--- +outline: deep +--- + +# Live Event Streams + +`void/live` provides Durable Object-backed fanout over Server-Sent Events. Use it +when one browser tab needs a single SSE connection that can subscribe and +unsubscribe from many application topics over time. + +Use `void/sse` for request-owned streams where the producer lives inside the same +route handler. Use `void/live` when later requests, mutations, scheduled jobs, or +queue consumers need to publish to clients that are already connected. + +## Define A Stream + +Create a server-only live stream descriptor: + +```ts +// src/live.ts +import { defineLiveStream } from 'void/live'; + +export const live = defineLiveStream({ + id: 'app', + allowAnonymousControl: true, +}); +``` + +Expose it from a normal route: + +```ts +// routes/live.ts +import { defineHandler } from 'void'; +import { live } from '../src/live'; + +export const GET = defineHandler((c) => live.connect(c)); +export const POST = defineHandler((c) => live.control(c)); +``` + +`GET` opens the SSE connection. `POST` accepts subscribe and unsubscribe control +operations for that connection. + +## Subscribe From The Browser + +Use the browser helper from `void/live/client`: + +```ts +import { connectLiveStream } from 'void/live/client'; + +const stream = connectLiveStream('/live', { + withCredentials: true, + retryDelay: 1_000, + onError(error) { + console.error(error); + }, +}); + +const unsubscribePost = await stream.subscribe({ + id: 'post-card', + topic: 'post:12', + onEvent(event) { + if (event.type === 'updated') { + console.log(event.data); + } + }, +}); + +const unsubscribeComments = await stream.subscribe({ + id: 'comments', + topic: 'comment:2323', +}); + +await unsubscribeComments(); +await unsubscribePost(); +stream.close(); +``` + +The helper opens one native `EventSource`, waits for it to open, then sends +batched `POST` control operations. Subscription ids are scoped to the +connection. Reusing an id replaces the previous subscription with that id. If +the transport drops, the helper reconnects and resubmits active subscriptions +with their latest `eventId` as `lastEventId`. + +## Publish + +Publish from any server-side code that has a Void runtime env: + +```ts +import { live } from '../src/live'; + +await live.publish( + 'post:12', + { title: 'Updated title' }, + { type: 'updated', eventId: 'post-12-v8' }, +); +``` + +Topics are opaque strings. Payloads are application-owned JSON. `type` and +`eventId` are copied into the JSON envelope; `void/live` does not interpret +them, persist them, or use native SSE `id` fields. + +If you are outside an active request/runtime context, pass env explicitly: + +```ts +await live.withEnv(env).publish('post:12', { title: 'Updated title' }); +``` + +## Authorization + +Every live connection has an owner. The owner is the identity allowed to send +control operations for that SSE connection. `connect()` resolves and stores the +owner; `control()` resolves the owner again and rejects the request if it does +not match. + +The quickstart uses an anonymous stream: + +```ts +export const live = defineLiveStream({ + id: 'app', + allowAnonymousControl: true, +}); +``` + +For authenticated streams, pass the same owner key to `connect()` and +`control()`: + +```ts +return live.connect(c, { owner: `user:${session.userId}` }); +return live.control(c, { owner: `user:${session.userId}` }); +``` + +If every route should derive ownership the same way, define it once with +`identifyConnection`: + +```ts +export const live = defineLiveStream({ + id: 'app', + async identifyConnection(ctx) { + const session = await getSession(ctx.request); + return session ? `user:${session.userId}` : null; + }, +}); +``` + +Then the route handlers can stay thin: + +```ts +export const GET = defineHandler((c) => live.connect(c)); +export const POST = defineHandler((c) => live.control(c)); +``` + +When `identifyConnection` returns `null` or `undefined`, Void falls back to the +current authenticated user and uses `user:${user.id}` when the user has a string +`id`. If neither path produces an owner and `allowAnonymousControl` is not set, +the request is rejected with `403`. + +For stream-local subscription rules, use `onSubscribe`: + +```ts +export const live = defineLiveStream({ + id: 'app', + async onSubscribe(ctx) { + const match = /^post:(.+)$/.exec(ctx.topic); + if (match && !(await canReadPost(ctx.env, ctx.user, match[1]))) { + return new Response('Forbidden', { status: 403 }); + } + }, +}); +``` + +## Limits + +`void/live` is designed for small and medium fanout. By default, a stream allows: + +- `256` active subscriptions per browser connection +- `256` active subscriptions per topic +- `100` subscribe or unsubscribe operations per control request +- `100` queued events per connection +- `64 KiB` per encoded event envelope + +A topic is the unit of fanout. For example, if each blog post uses a topic like +`post:${postId}`, then each post can have up to `256` active live subscribers at +one time. Other posts use separate topics and have separate limits. + +Streams can have many possible topics. Topics are created on demand when clients +subscribe or publishers send events, so an app with thousands of posts, rooms, or +documents can use one topic per entity without provisioning them ahead of time. +The limit applies to each active topic, not to the total number of topic names +your app might use. + +For example, a blog with `10,000` posts can model its live capacity as +`10,000 posts × up to 256 active subscribers per post topic`. + +That shape works because each post topic is independent. A single post with more +than `256` active live subscribers would need a larger broadcast design. + +These limits apply to live subscriptions, not total users or total page views. +One user with two open tabs may count twice. Disconnected subscriptions are +pruned, but they can count until the runtime observes that they are stale. + +You can lower limits per stream: + +```ts +export const live = defineLiveStream({ + id: 'app', + limits: { + maxSubscriptionsPerTopic: 64, + }, +}); +``` + +`maxSubscriptionsPerTopic` cannot be raised above `256`. For larger broadcast +workloads, shard topics in userland or use a dedicated realtime system. + +## Delivery Semantics + +`void/live` is an at-most-once live fanout primitive: + +- one SSE connection can hold many topic subscriptions +- publish cost is proportional to subscribers of the topic +- events are ordered within one topic +- clients using the raw HTTP protocol must resubscribe after reconnect +- deploys, rollbacks, Worker restarts, browser reconnects, and network changes + can drop live state +- durable replay, cache invalidation, live queries, and client state management + belong in userland + +Use an application database, queue, or custom Durable Object for replay. Use +`eventId` and caller-provided `lastEventId` values as application protocol +fields when you build that layer. diff --git a/docs/reference/api.md b/docs/reference/api.md index 59f55ca71..6e7f8d0af 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -904,6 +904,8 @@ Solid `Link` GET `data` is merged into the rendered `href` query string. Primiti | `void/ws` | `defineRoom`, `defineWebSocket`, `connect`, WebSocket context and connection types | | `void/sse` | `eventStream`, `formatSse`, `formatSseText`, `getLastEventId`, `SseStreamClosedError`, SSE message and stream types | | `void/sse/client` | `connectEventStream`, browser `EventSource` wrapper types | +| `void/live` | `defineLiveStream`, SSE topic fanout runtime types. Server-only. | +| `void/live/client` | `connectLiveStream`, browser helper for one SSE connection plus POST subscribe/unsubscribe control. | | `void/response` | `convertReturnValue` | | `void/validator` | `runValidation`, `ValidatorSlots`, `HandlerInput` | | `void/drizzle-zod` | Re-exports [`drizzle-zod`](https://orm.drizzle.team/docs/zod) for schema-derived Zod validators for Drizzle tables | diff --git a/packages/void/package.json b/packages/void/package.json index c99b02ca9..c6891919e 100644 --- a/packages/void/package.json +++ b/packages/void/package.json @@ -33,6 +33,9 @@ "#self/sandbox": "./dist/runtime/sandbox.mjs", "#self/sse": "./dist/runtime/sse.mjs", "#self/sse/client": "./dist/runtime/sse-client.mjs", + "#self/live": "./dist/runtime/live.mjs", + "#self/live/client": "./dist/runtime/live-client.mjs", + "#self/runtime/live-server": "./dist/runtime/live-server.mjs", "#self/runtime/fetch": "./dist/runtime/fetch.mjs", "#self/runtime/fetch-stream": "./dist/runtime/fetch-stream.mjs", "#self/runtime/migration-handler": "./dist/runtime/migration-handler.mjs", @@ -98,6 +101,16 @@ "import": "./dist/runtime/sse-client.mjs", "require": "./dist/runtime/sse-client.mjs" }, + "./live": { + "types": "./dist/runtime/live.d.mts", + "import": "./dist/runtime/live.mjs", + "require": "./dist/runtime/live.mjs" + }, + "./live/client": { + "types": "./dist/runtime/live-client.d.mts", + "import": "./dist/runtime/live-client.mjs", + "require": "./dist/runtime/live-client.mjs" + }, "./ws": { "types": "./dist/runtime/ws.d.mts", "import": "./dist/runtime/ws.mjs", diff --git a/packages/void/src/cli/deploy.ts b/packages/void/src/cli/deploy.ts index c562f7443..048193b3f 100644 --- a/packages/void/src/cli/deploy.ts +++ b/packages/void/src/cli/deploy.ts @@ -51,7 +51,12 @@ import { import { resolveRemoteMode, stripInternalEnvKeys } from '../shared/env.ts'; import { validateSsrEntry } from '../ssr/entry.ts'; import { getToken, getTokenSource } from './auth.ts'; -import { isExpiredTokenError, PlatformClient, type ManifestOnlyDeployPayload } from './client.ts'; +import { + isExpiredTokenError, + PlatformClient, + type DeploymentResponse, + type ManifestOnlyDeployPayload, +} from './client.ts'; import { promptAndCreateProject } from './create-project.ts'; import { formatConnectionLostMessage, @@ -98,7 +103,14 @@ import type { DevRedirectRule, HeaderRule, } from './types.ts'; -import { readWranglerCompat, type WranglerCompat } from './wrangler.ts'; +import { + readExistingLiveBindingConfig, + readWranglerCompat, + resolveLiveBindingConfig, + syncWranglerBindings, + type LiveBindingConfig, + type WranglerCompat, +} from './wrangler.ts'; import { needsWrapper, generateWrapper, bundleEntry, cleanupWrapper } from './wrapper.ts'; class DeployEventError extends Error {} @@ -119,10 +131,95 @@ interface ResourceSummary { type?: string; } +const EMPTY_BINDINGS = { + needsD1: false, + needsKV: false, + needsR2: false, + needsAuth: false, + needsQueues: false, + needsAI: false, + needsSandbox: false, +}; + function getErrorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } +function readLiveBindingConfigFromDeployment( + deployment: DeploymentResponse | undefined, +): LiveBindingConfig | undefined { + if (!deployment?.manifest) { + return undefined; + } + let manifest: DeployManifest; + try { + manifest = JSON.parse(deployment.manifest) as DeployManifest; + } catch { + return undefined; + } + if ( + typeof manifest.live?.bindingName !== 'string' || + typeof manifest.live.className !== 'string' + ) { + return undefined; + } + return { + bindingName: manifest.live.bindingName, + className: manifest.live.className, + }; +} + +export async function ensurePreviouslyDeployedLiveBindingConfig( + root: string, + client: PlatformClient, + projectId: string, + cliLog?: CliLogger, +): Promise { + if (readExistingLiveBindingConfig(root)) { + return false; + } + + const { deployments, activeDeploymentId } = await client.listDeployments(projectId); + const live = readLiveBindingConfigFromDeployment( + deployments.find((deployment) => deployment.id === activeDeploymentId), + ); + if (!live) { + return false; + } + + const synced = syncWranglerBindings(root, { ...EMPTY_BINDINGS, needsLive: true }, { live }); + if (synced) { + cliLog?.info('live_binding_restored', { + bindingName: live.bindingName, + className: live.className, + }); + } + return synced; +} + +export async function resolveDeployLiveBindingConfig( + root: string, + client: PlatformClient, + projectId: string, + opts: { + isFrameworkMode: boolean; + needsLive: boolean | undefined; + webSockets: Array<{ bindingName: string; className: string }>; + sandboxClassName?: string; + cliLog?: CliLogger; + }, +): Promise { + if (opts.isFrameworkMode || !opts.needsLive) { + return undefined; + } + + await ensurePreviouslyDeployedLiveBindingConfig(root, client, projectId, opts.cliLog); + return resolveLiveBindingConfig(root, { + reservedBindings: opts.webSockets.map((route) => route.bindingName), + reservedClasses: [...opts.webSockets.map((route) => route.className), opts.sandboxClassName], + }); +} + /** * Resolve the Cloudflare Workers Static Assets `not_found_handling` config. * @@ -981,6 +1078,10 @@ async function runFrameworkDeploy( ); process.exit(1); } + if (bindings.needsLive) { + log.error('deploy: void/live is currently supported in Void apps only.'); + process.exit(1); + } const migrationsDir = join(root, 'db', 'migrations'); @@ -1307,6 +1408,10 @@ async function runFullDeploy( log.error('deploy: Cloudflare Sandboxes are currently supported in Void apps only.'); process.exit(1); } + if (isFrameworkMode && bindings.needsLive) { + log.error('deploy: void/live is currently supported in Void apps only.'); + process.exit(1); + } const sandboxConfig = bindings.needsSandbox ? resolveSandboxConfig(deployConfig, root) @@ -1400,6 +1505,13 @@ async function runFullDeploy( ); const routeScan = await scanRoutes(root, { output: deployConfig.output }); const webSockets = routeScan.websockets ?? scanWebSocketRoutesSync(root); + const liveConfig = await resolveDeployLiveBindingConfig(root, client, config.projectId, { + isFrameworkMode, + needsLive: effectiveBindings.needsLive, + webSockets, + sandboxClassName: sandboxConfig?.className, + cliLog, + }); const queues = scanQueuesSync(root).map((q) => ({ name: q.name, bindingName: q.bindingName, @@ -1455,7 +1567,8 @@ async function runFullDeploy( const allPrerenderable = pageScan.pages.every( (p) => p.prerender && ((p.params.length === 0 && !p.catchAll) || p.hasGetPrerenderPaths), ); - const hasNonPageBackend = hasRoutes || hasMiddleware || hasWebSockets || authEnabled; + const hasNonPageBackend = + hasRoutes || hasMiddleware || hasWebSockets || authEnabled || liveConfig !== undefined; if (allPrerenderable && !hasNonPageBackend && schedules.length === 0 && queues.length === 0) { log.info('All pages prerendered — deploying as static site'); cliLog?.info('full_to_static_redirect', { reason: 'all_pages_prerendered' }); @@ -1573,6 +1686,7 @@ async function runFullDeploy( compatibilityDate: wranglerCompat?.compatibilityDate, compatibilityFlags: wranglerCompat?.compatibilityFlags, ...(sandboxConfig && { sandbox: sandboxConfig }), + ...(liveConfig && { live: liveConfig }), revalidateQueryAllowlist: deployConfig.routing?.revalidateQueryAllowlist, }; @@ -1585,7 +1699,7 @@ async function runFullDeploy( } const manifest = buildVoidManifest({ - bindings: effectiveBindings, + bindings: liveConfig ? { ...effectiveBindings, needsLive: true } : effectiveBindings, migrations: validatedMigrations, schedules, ssr: isSsr || hasRoutes || hasMiddleware || hasWebSockets || authEnabled, diff --git a/packages/void/src/cli/package.ts b/packages/void/src/cli/package.ts index 3327aca3e..0ca7da1d4 100644 --- a/packages/void/src/cli/package.ts +++ b/packages/void/src/cli/package.ts @@ -19,11 +19,20 @@ import type { ScheduleMeta, WebSocketManifestMeta, } from './types.ts'; +import { + DEFAULT_LIVE_BINDING, + DEFAULT_LIVE_CLASS, + LIVE_MIGRATION_TAG, + type LiveBindingConfig, +} from './wrangler.ts'; const textEncoder = new TextEncoder(); type ProgressCallback = (message: string) => void; +const FRAMEWORK_LIVE_UNSUPPORTED_MESSAGE = + 'deploy: void/live is currently supported in Void apps only.'; + /** * Strip dev-only / internal-only fields from each rule before shipping the * manifest. The resulting entries contain only `{ source, destination, status }`. @@ -383,6 +392,8 @@ export interface PackageOptions { compatibilityFlags?: Array; /** Cloudflare Sandbox container metadata for platform deploys. */ sandbox?: ResolvedSandboxConfig; + /** Resolved void/live Durable Object binding and class names. */ + live?: LiveBindingConfig; /** Query params that participate in ISR variant keys for dispatch rewrites. */ revalidateQueryAllowlist?: Record>; } @@ -416,6 +427,18 @@ function toSandboxManifest(sandbox: ResolvedSandboxConfig): SandboxManifestMeta }; } +function toLiveManifest(live?: LiveBindingConfig): import('./types.ts').LiveManifestMeta { + const bindingName = live?.bindingName ?? DEFAULT_LIVE_BINDING; + const className = live?.className ?? DEFAULT_LIVE_CLASS; + return { + bindingName, + className, + storage: 'sqlite', + classes: [className], + migrations: [{ tag: LIVE_MIGRATION_TAG, newSqliteClasses: [className] }], + }; +} + /** * If the user configured `routing.fallbacks` on a non-SPA app, the platform * still applies the rules but no `/* -> /index.html` SPA shell is synthesised @@ -517,6 +540,9 @@ export function buildVoidManifest(opts: { binding: options.bindingNames?.sandbox ?? options.sandbox.binding, }); } + if (bindings.needsLive) { + manifest.live = toLiveManifest(options?.live); + } if (migrations && migrations.length > 0) { manifest.migrations = { @@ -731,6 +757,9 @@ export function buildFrameworkManifest(opts: { if (opts.bindings.needsAI) { manifest.bindings.ai = true; } + if (opts.bindings.needsLive) { + throw new Error(FRAMEWORK_LIVE_UNSUPPORTED_MESSAGE); + } if (opts.migrations && opts.migrations.length > 0) { manifest.migrations = { diff --git a/packages/void/src/cli/types.ts b/packages/void/src/cli/types.ts index 0ad9a8f53..8be520f4d 100644 --- a/packages/void/src/cli/types.ts +++ b/packages/void/src/cli/types.ts @@ -130,6 +130,19 @@ export interface SandboxManifestMeta { migrations?: Array; } +export interface LiveMigrationMeta { + tag: string; + newSqliteClasses?: Array; +} + +export interface LiveManifestMeta { + bindingName: string; + className: string; + storage: 'sqlite'; + classes: Array; + migrations?: Array; +} + export interface DeployManifest { version: number; /** @@ -176,6 +189,7 @@ export interface DeployManifest { fallbackRules?: Array; websocket?: WebSocketManifestMeta; sandbox?: SandboxManifestMeta; + live?: LiveManifestMeta; } export type DeployEvent = diff --git a/packages/void/src/cli/wrangler.ts b/packages/void/src/cli/wrangler.ts index 98bf0693d..44942f4d6 100644 --- a/packages/void/src/cli/wrangler.ts +++ b/packages/void/src/cli/wrangler.ts @@ -1,3 +1,4 @@ +import { createHash } from 'node:crypto'; import { readFileSync, writeFileSync, existsSync } from 'node:fs'; import { parse, type ParseError } from 'jsonc-parser'; import { join } from 'pathe'; @@ -14,6 +15,20 @@ import { const COMPAT_DATE_RE = /^\d{4}-\d{2}-\d{2}$/; const NODEJS_ALS_FLAG = 'nodejs_als'; const VOID_CONFIG_FILE = 'void.json'; +export const DEFAULT_LIVE_BINDING = 'VOID_LIVE'; +export const DEFAULT_LIVE_CLASS = 'VoidLiveStreamDurableObject'; +export const LIVE_MIGRATION_TAG = 'live-v1'; + +export interface LiveBindingConfig { + bindingName: string; + className: string; +} + +type DurableObjectMigration = { + tag?: string; + new_classes?: Array; + new_sqlite_classes?: Array; +}; /** Latest compatibility date the Void test suite is pinned against. */ export const LATEST_KNOWN_COMPAT_DATE = '2026-02-24'; @@ -24,6 +39,130 @@ export interface WranglerCompat { compatibilityFlags?: Array; } +export function hasDurableObjectSqliteMigration( + migrations: Array, + className: string, +): boolean { + return migrations.some((migration) => migration.new_sqlite_classes?.includes(className)); +} + +export function nextDurableObjectMigrationTag( + migrations: Array, + baseTag: string, +): string { + const existingTags = new Set(migrations.map((migration) => migration.tag).filter(Boolean)); + if (!existingTags.has(baseTag)) { + return baseTag; + } + let index = 2; + while (existingTags.has(`${baseTag}-${index}`)) { + index++; + } + return `${baseTag}-${index}`; +} + +export function resolveLiveBindingConfig( + root: string, + options?: { + existingBindings?: Array<{ name: string; class_name: string }>; + reservedBindings?: Array; + reservedClasses?: Array; + }, +): LiveBindingConfig { + const durableObjectBindings = [ + ...readWranglerDurableObjectBindings(root), + ...(options?.existingBindings ?? []), + ]; + + const generated = findGeneratedLiveBinding(durableObjectBindings); + if (generated) { + return { bindingName: generated.name, className: generated.class_name }; + } + + const usedBindings = new Set([ + ...durableObjectBindings.map((binding) => binding.name), + ...(options?.reservedBindings ?? []).filter((name): name is string => typeof name === 'string'), + ]); + const usedClasses = new Set([ + ...durableObjectBindings.map((binding) => binding.class_name), + ...(options?.reservedClasses ?? []).filter((name): name is string => typeof name === 'string'), + ]); + + return { + bindingName: resolveGeneratedName(DEFAULT_LIVE_BINDING, usedBindings, 'binding'), + className: resolveGeneratedName(DEFAULT_LIVE_CLASS, usedClasses, 'class'), + }; +} + +export function readExistingLiveBindingConfig(root: string): LiveBindingConfig | undefined { + const generated = findGeneratedLiveBinding(readWranglerDurableObjectBindings(root)); + return generated ? { bindingName: generated.name, className: generated.class_name } : undefined; +} + +function readWranglerDurableObjectBindings( + root: string, +): Array<{ name: string; class_name: string }> { + for (const name of ['wrangler.jsonc', 'wrangler.json']) { + try { + const parsed = parseJsonc(readFileSync(join(root, name), 'utf-8')); + const bindings = ( + parsed.durable_objects as + | { bindings?: Array<{ name?: unknown; class_name?: unknown }> } + | undefined + )?.bindings; + if (!Array.isArray(bindings)) { + return []; + } + return bindings.filter( + (binding): binding is { name: string; class_name: string } => + typeof binding.name === 'string' && typeof binding.class_name === 'string', + ); + } catch { + // Try the next wrangler config filename. + } + } + return []; +} + +function isGeneratedLiveName(name: string, base: string): boolean { + return name === base || name.match(generatedLiveNamePattern(base)) !== null; +} + +function generatedLiveNamePattern(base: string): RegExp { + return new RegExp(`^${escapeRegExp(base)}_[0-9a-f]{8}$`); +} + +function escapeRegExp(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); +} + +function findGeneratedLiveBinding( + bindings: Array<{ name: string; class_name: string }>, +): { name: string; class_name: string } | undefined { + return bindings.find( + (binding) => + isGeneratedLiveName(binding.name, DEFAULT_LIVE_BINDING) && + isGeneratedLiveName(binding.class_name, DEFAULT_LIVE_CLASS), + ); +} + +function resolveGeneratedName(base: string, used: Set, kind: string): string { + if (!used.has(base)) { + return base; + } + + for (let attempt = 0; ; attempt++) { + const suffix = createHash('sha256') + .update(`void/live:${kind}:${base}:${attempt}`) + .digest('hex') + .slice(0, 8); + const candidate = `${base}_${suffix}`; + if (!used.has(candidate)) { + return candidate; + } + } +} + /** * Read compatibility settings from `wrangler.jsonc`/`wrangler.json` and * `void.json`, with priority: void.json > wrangler.jsonc > build output. @@ -183,6 +322,7 @@ export function syncWranglerBindings( bindingNames?: { d1?: string; kv?: string; r2?: string; sandbox?: string }; websockets?: Array; sandbox?: ResolvedSandboxConfig; + live?: boolean | LiveBindingConfig; }, ): boolean { // Find the wrangler config file @@ -336,6 +476,53 @@ export function syncWranglerBindings( } } + if (bindings.needsLive || options?.live) { + const existingDurableObjectBindings = + (( + parsed.durable_objects as + | { bindings?: Array<{ name: string; class_name: string }> } + | undefined + )?.bindings as Array<{ name: string; class_name: string }> | undefined) ?? []; + const liveConfig = + typeof options?.live === 'object' + ? options.live + : resolveLiveBindingConfig(root, { + existingBindings: existingDurableObjectBindings, + reservedBindings: options?.websockets?.map((route) => route.bindingName), + reservedClasses: [ + ...(options?.websockets?.map((route) => route.className) ?? []), + options?.sandbox?.className, + ], + }); + if (!existingDurableObjectBindings.some((binding) => binding.name === liveConfig.bindingName)) { + parsed.durable_objects = { + bindings: [ + ...existingDurableObjectBindings, + { + name: liveConfig.bindingName, + class_name: liveConfig.className, + }, + ], + }; + modified = true; + } + + const migrations = Array.isArray(parsed.migrations) + ? (parsed.migrations as Array) + : []; + const hasMigration = hasDurableObjectSqliteMigration(migrations, liveConfig.className); + if (!hasMigration) { + parsed.migrations = [ + ...migrations, + { + tag: nextDurableObjectMigrationTag(migrations, LIVE_MIGRATION_TAG), + new_sqlite_classes: [liveConfig.className], + }, + ]; + modified = true; + } + } + if (created || modified) { writeFileSync(configPath, JSON.stringify(parsed, null, 2) + '\n'); } diff --git a/packages/void/src/index.ts b/packages/void/src/index.ts index 65d99e8e1..8c84a02b2 100644 --- a/packages/void/src/index.ts +++ b/packages/void/src/index.ts @@ -18,7 +18,12 @@ import { readProjectConfig } from './cli/project.ts'; import { ensureWranglerCompatibilityDate, ensureWranglerNodejsAlsFlag, + hasDurableObjectSqliteMigration, + LIVE_MIGRATION_TAG, + nextDurableObjectMigrationTag, readWranglerCompat, + readExistingLiveBindingConfig, + resolveLiveBindingConfig, syncWranglerBindings, } from './cli/wrangler.ts'; import { readConfig, resolveBindingNames, isNodeTarget, getDatabaseDialect } from './config.ts'; @@ -39,6 +44,7 @@ import { writeFrameworkTriggersDispatch, } from './plugin-framework-triggers.ts'; import { + DEFAULT_SCAN_DIRS, inferProjectBindings, detectFramework, FRAMEWORK_SCAN_DIRS, @@ -567,6 +573,9 @@ export function voidPlugin(options?: { if (detected && bindings.needsSandbox) { throw new Error('sandbox: Cloudflare Sandboxes are currently supported in Void apps only.'); } + if (detected && bindings.needsLive) { + throw new Error('live: void/live is currently supported in Void apps only.'); + } // Server-only void/* imports (db, kv, ai, etc.) must not pull in // cloudflare:workers or node: builtins when bundled for the browser. @@ -580,6 +589,7 @@ export function voidPlugin(options?: { 'void/storage': 'export const storage = {}; export const createStorage = () => ({});', 'void/isr': 'export const revalidate = async () => {};', 'void/sandbox': 'export const sandbox = {}; export const getSandbox = () => ({});', + 'void/live': '', }; const STUB_RESOLVE_FILTER = new RegExp( `^(${Object.keys(SERVER_ONLY_STUBS).join('|').replaceAll('/', '\\/')})$`, @@ -594,6 +604,11 @@ export function voidPlugin(options?: { filter: { id: STUB_RESOLVE_FILTER }, handler(id) { if (this.environment?.name === 'client') { + if (id === 'void/live') { + this.error( + 'void/live is server-only and cannot be imported by browser code. Import void/live/client instead.', + ); + } return STUB_PREFIX + id; } }, @@ -610,6 +625,17 @@ export function voidPlugin(options?: { // branches can safely close over these bindings. const queues = framework ? [] : scanQueuesSync(root); const websocketRoutes = framework ? [] : scanWebSocketRoutesSync(root); + const liveConfig = !isNodeTarget(config.target) + ? bindings.needsLive + ? resolveLiveBindingConfig(root, { + reservedBindings: websocketRoutes.map((route) => route.bindingName), + reservedClasses: [ + ...websocketRoutes.map((route) => route.className), + sandboxConfig?.className, + ], + }) + : readExistingLiveBindingConfig(root) + : undefined; // Class B/C: lighter composition — DB types + migrations only. // The framework's own adapter handles the CF worker build and dev server. @@ -655,6 +681,7 @@ export function voidPlugin(options?: { bindingNames, websockets: websocketRoutes, sandbox: sandboxConfig, + live: liveConfig, }); if (synced) { voidLog('Synced bindings to wrangler.jsonc'); @@ -936,6 +963,53 @@ export function voidPlugin(options?: { } } + if (liveConfig) { + const currentDurableObjects = + ((result as Record).durable_objects as + | { bindings?: Array<{ name: string; class_name: string }> } + | undefined) ?? + ((resolved as Record).durable_objects as + | { bindings?: Array<{ name: string; class_name: string }> } + | undefined); + const existingDurableObjectBindings = + (currentDurableObjects?.bindings as + | Array<{ name: string; class_name: string }> + | undefined) ?? []; + if ( + !existingDurableObjectBindings.some((binding) => binding.name === liveConfig.bindingName) + ) { + (result as Record).durable_objects = { + bindings: [ + ...existingDurableObjectBindings, + { + name: liveConfig.bindingName, + class_name: liveConfig.className, + }, + ], + }; + } + + const currentMigrations = + (result as Record).migrations ?? + (resolved as Record).migrations; + const existingMigrations = Array.isArray(currentMigrations) + ? (currentMigrations as Array<{ + tag?: string; + new_classes?: Array; + new_sqlite_classes?: Array; + }>) + : []; + if (!hasDurableObjectSqliteMigration(existingMigrations, liveConfig.className)) { + (result as Record).migrations = [ + ...existingMigrations, + { + tag: nextDurableObjectMigrationTag(existingMigrations, LIVE_MIGRATION_TAG), + new_sqlite_classes: [liveConfig.className], + }, + ]; + } + } + // AI — in local dev, inject VC credentials for HTTPS proxy path. // In production, the platform adds a __VOID_PROXY service binding instead. if (viteCommand === 'serve' && bindings.needsAI) { @@ -1124,7 +1198,7 @@ export function voidPlugin(options?: { if (!autoDetectBindings && !autoDetectAuth) { return; } - const effectiveDirs = scanDirs ?? ['routes', 'middleware', 'queues', 'pages', 'crons']; + const effectiveDirs = scanDirs ?? DEFAULT_SCAN_DIRS; const watchDirs = [...new Set([...effectiveDirs, 'src'])].map((d) => join(root, d)); let hintShown = false; const check = (path: string) => { @@ -1284,6 +1358,11 @@ export function voidPlugin(options?: { `sandbox: Cloudflare Sandboxes require target: "cloudflare". They are not available with target: '${config.target}'.`, ); } + if (bindings.needsLive) { + throw new Error( + `live: void/live requires target: "cloudflare". It is not available with target: '${config.target}'.`, + ); + } if (config.routing?.revalidate) { voidWarn( '"routing.revalidate" in void.json is ignored with target: "' + @@ -1383,6 +1462,7 @@ export function voidPlugin(options?: { hasSandbox: bindings.needsSandbox, sandboxBindingName: sandboxConfig?.binding, sandboxClassName: sandboxConfig?.className, + live: liveConfig, }), drizzlePlugin(root, true, dialect, { forceVirtualDb: authEnabled }), migrationPlugin(root, dialect, 'local', join(root, '.void'), { diff --git a/packages/void/src/inference/walk.ts b/packages/void/src/inference/walk.ts index 308d20332..6f25ff8c5 100644 --- a/packages/void/src/inference/walk.ts +++ b/packages/void/src/inference/walk.ts @@ -11,6 +11,7 @@ export interface InferredBindings { needsQueues: boolean; needsAI: boolean; needsSandbox: boolean; + needsLive?: boolean; } const ENV_BINDING_RE = /\benv\s*\.\s*(DB|KV|STORAGE|QUEUE_\w+|AI|SANDBOX)\b/; @@ -32,6 +33,7 @@ export function inferBindingsFromSource(code: string, filename: string): Inferre needsQueues: false, needsAI: false, needsSandbox: false, + needsLive: false, }; // Tier 1: Fast import detection via es-module-lexer @@ -65,6 +67,11 @@ export function inferBindingsFromSource(code: string, filename: string): Inferre case 'void/sandbox': result.needsSandbox = true; break; + case 'void/live': + if (!isTypeOnlyImportStatement(code.slice(imp.ss, imp.se + 1))) { + result.needsLive = true; + } + break; case 'void/client': { // Check if the `auth` specifier is imported (not just `fetch`) const stmt = code.slice(imp.ss, imp.se + 1); @@ -98,13 +105,23 @@ export function inferBindingsFromSource(code: string, filename: string): Inferre if (/\bfrom\s+["']void\/sandbox["']/.test(code)) { result.needsSandbox = true; } + if (hasRuntimeImportFrom(code, 'void/live')) { + result.needsLive = true; + } if (/\bfrom\s+["']void\/client["']/.test(code) && /\bauth\b/.test(code)) { result.needsAuth = true; } } // If imports resolved all bindings, we're done - if (result.needsD1 && result.needsKV && result.needsR2 && result.needsAI && result.needsSandbox) { + if ( + result.needsD1 && + result.needsKV && + result.needsR2 && + result.needsAI && + result.needsSandbox && + result.needsLive + ) { return result; } @@ -175,6 +192,45 @@ export function inferBindingsFromSource(code: string, filename: string): Inferre return result; } +function isTypeOnlyImportStatement(statement: string): boolean { + if (/^\s*import\s+type\b/.test(statement)) { + return true; + } + + const specifierStart = statement.indexOf('{'); + const specifierEnd = statement.lastIndexOf('}'); + if (specifierStart === -1 || specifierEnd === -1 || specifierEnd < specifierStart) { + return false; + } + + const beforeSpecifiers = statement.slice(0, specifierStart); + if (!/^\s*import\s*$/.test(beforeSpecifiers)) { + return false; + } + + const specifiers = statement + .slice(specifierStart + 1, specifierEnd) + .split(',') + .map((specifier) => specifier.trim()) + .filter(Boolean); + return specifiers.length > 0 && specifiers.every((specifier) => specifier.startsWith('type ')); +} + +function hasRuntimeImportFrom(code: string, source: string): boolean { + const escapedSource = source.replaceAll('/', '\\/'); + const importRe = new RegExp( + `\\bimport\\s+(?:[\\s\\S]*?\\bfrom\\s+)?["']${escapedSource}["']\\s*;?`, + 'g', + ); + for (const match of code.matchAll(importRe)) { + const statement = match[0]; + if (!isTypeOnlyImportStatement(statement)) { + return true; + } + } + return false; +} + /** * Merge multiple InferredBindings into a single result (union). */ @@ -187,5 +243,6 @@ export function mergeBindings(bindings: Array): InferredBindin needsQueues: bindings.some((b) => b.needsQueues), needsAI: bindings.some((b) => b.needsAI), needsSandbox: bindings.some((b) => b.needsSandbox), + needsLive: bindings.some((b) => b.needsLive), }; } diff --git a/packages/void/src/plugin-inference.ts b/packages/void/src/plugin-inference.ts index b28010acb..4841b0574 100644 --- a/packages/void/src/plugin-inference.ts +++ b/packages/void/src/plugin-inference.ts @@ -28,6 +28,16 @@ export const FRAMEWORK_SCAN_DIRS = [ 'queues', ] as const; +/** Directories to scan for binding inference in Void app mode. */ +export const DEFAULT_SCAN_DIRS = [ + 'routes', + 'middleware', + 'queues', + 'pages', + 'crons', + 'src', +] as const; + /** * Map from npm package name to framework identifier. * Only packages that unambiguously indicate framework mode are listed. @@ -121,10 +131,11 @@ export function inferProjectBindings( needsQueues: false, needsAI: Boolean(options.bindings.ai), needsSandbox: false, + needsLive: false, }; } - const dirs = scanDirs ?? ['routes', 'middleware', 'queues', 'pages', 'crons']; + const dirs = scanDirs ?? DEFAULT_SCAN_DIRS; const allBindings: Array = []; for (const dir of dirs) { @@ -149,7 +160,8 @@ export function inferProjectBindings( allBindings.push(inferBindingsFromSource(code, filename)); } - // Also scan src/ for auth imports (unless src/ is already in scanDirs) + // Custom scan directories should still see client/auth helpers and colocated + // server-only modules under src/. const dirsSet = new Set(dirs); if (!dirsSet.has('src')) { const srcPath = join(root, 'src'); diff --git a/packages/void/src/plugin-routing.ts b/packages/void/src/plugin-routing.ts index 8e8aee68f..18cc7f8c0 100644 --- a/packages/void/src/plugin-routing.ts +++ b/packages/void/src/plugin-routing.ts @@ -70,6 +70,7 @@ interface RoutingPluginOptions { hasSandbox?: boolean; sandboxBindingName?: string; sandboxClassName?: string; + live?: { bindingName: string; className: string }; } export function routingPlugin(root: string, options: RoutingPluginOptions): Plugin { @@ -158,6 +159,8 @@ export function routingPlugin(root: string, options: RoutingPluginOptions): Plug projectRoot: root, homeDir: homedir(), webSocketRuntimePath: resolveFile('#self/runtime/ws-server'), + liveRuntimePath: resolveFile('#self/runtime/live-server'), + hasLive: options.live !== undefined, reactRefresh: false, dev: false, devTriggerToken: options.devTriggerToken, @@ -338,6 +341,8 @@ export function routingPlugin(root: string, options: RoutingPluginOptions): Plug runtimePath: sandboxRuntimePath, } : false, + undefined, + options.live, ), ); } @@ -450,6 +455,7 @@ export function routingPlugin(root: string, options: RoutingPluginOptions): Plug } : false, shouldServeSpaFallback(), + options.live, ), ); } @@ -568,6 +574,7 @@ export function routingPlugin(root: string, options: RoutingPluginOptions): Plug } : false, shouldServeSpaFallback(), + options.live, ), ); } diff --git a/packages/void/src/router/compile.ts b/packages/void/src/router/compile.ts index a348b858f..baf057ea9 100644 --- a/packages/void/src/router/compile.ts +++ b/packages/void/src/router/compile.ts @@ -673,6 +673,10 @@ export interface CompileOptions { explicitPrerenderPaths?: Array; /** Absolute path to the internal WebSocket runtime module */ webSocketRuntimePath?: string; + /** Absolute path to the internal Live runtime module */ + liveRuntimePath?: string; + /** Whether the app uses void/live. */ + hasLive?: boolean; } export interface SsrCompileOptions { @@ -758,6 +762,11 @@ export function compileRouteTable( `import { createWebSocketDurableObject as __createWebSocketDurableObject, handleWebSocketUpgrade as __handleWebSocketUpgrade, forwardWebSocketRequest as __forwardWebSocketRequest } from ${JSON.stringify(options.webSocketRuntimePath)};`, ); } + if (options.hasLive && options.liveRuntimePath) { + lines.push( + `import { createLiveDurableObject as __createLiveDurableObject } from ${JSON.stringify(options.liveRuntimePath)};`, + ); + } lines.push(''); // Generate eager imports for middleware @@ -1156,6 +1165,11 @@ export function compileRouteTable( lines.push(''); } + if (options.hasLive) { + lines.push('export const VoidLiveStreamDurableObject = __createLiveDurableObject();'); + lines.push(''); + } + // Pages mode — generate page route handlers if (pages && scan.pages.pages.length > 0) { compilePageRoutes(lines, scan, pages, options); @@ -2153,6 +2167,7 @@ export function compileWorkerEntry( envSchemaModule?: { importPath: string; helperPath: string }, sandbox?: boolean | { bindingName?: string; className?: string; runtimePath?: string }, serveSpaFallback = false, + live?: boolean | { bindingName?: string; className?: string }, ): string { const remotePath = remote?.remotePath; const hasSandbox = sandbox === true || typeof sandbox === 'object'; @@ -2170,6 +2185,20 @@ export function compileWorkerEntry( ? `export { __VoidSandbox as Sandbox };\n__setSandboxBindingName(${JSON.stringify(sandboxBindingName)});\n` : `export class ${sandboxClassName} extends __VoidSandbox {}\n__setSandboxBindingName(${JSON.stringify(sandboxBindingName)});\n` : ''; + const hasLive = live === true || typeof live === 'object'; + const liveBindingName = + typeof live === 'object' ? (live.bindingName ?? 'VOID_LIVE') : 'VOID_LIVE'; + const liveClassName = + typeof live === 'object' + ? (live.className ?? 'VoidLiveStreamDurableObject') + : 'VoidLiveStreamDurableObject'; + const liveRouteImport = hasLive ? ', VoidLiveStreamDurableObject' : ''; + const liveRuntimeImport = hasLive ? `import { __setLiveBindingName } from "void/live";\n` : ''; + const liveExport = hasLive + ? liveClassName === 'VoidLiveStreamDurableObject' + ? `export { VoidLiveStreamDurableObject };\n__setLiveBindingName(${JSON.stringify(liveBindingName)});\n` + : `export { VoidLiveStreamDurableObject as ${liveClassName} };\n__setLiveBindingName(${JSON.stringify(liveBindingName)});\n` + : ''; const runtimeEnvImport = `import { withRuntimeEnv } from ${JSON.stringify(runtimeEnvPath)};\n`; const remoteImport = remote ? `import { createRemoteEnv } from ${JSON.stringify(remote.remotePath)};\n` @@ -2323,7 +2352,7 @@ ${envValidateCall} const userEnv = __filterInternalBindings(env); }`; if (hasQueues) { - return `${sandboxImport}import app, { scheduled, __voidCollectPrerenderPaths, __voidHandleWebSocket, __withRawEnv, __processQueue${webSocketClasses.length > 0 ? `, ${webSocketClasses.join(', ')}` : ''} } from 'virtual:void-routes'; + return `${sandboxImport}${liveRuntimeImport}import app, { scheduled, __voidCollectPrerenderPaths, __voidHandleWebSocket, __withRawEnv, __processQueue${webSocketClasses.length > 0 ? `, ${webSocketClasses.join(', ')}` : ''}${liveRouteImport} } from 'virtual:void-routes'; ${runtimeEnvImport}${envSchemaImport}${remoteImport}${filterBindingsHelper}${assetHelper} ${internalRouteHelper} ${scheduledHandler} @@ -2349,14 +2378,16 @@ ${envValidateCall}${remotePath ? ` if (env.__VOID_REMOTE === "1") env = createR } ${webSocketClasses.map((name) => `export { ${name} };`).join('\n')} +${liveExport} ${sandboxExport} export default { fetch: ${fetchHandler}, scheduled: handleScheduled, queue };`; } - return `${sandboxImport}import app, { scheduled, __voidCollectPrerenderPaths, __voidHandleWebSocket, __withRawEnv${webSocketClasses.length > 0 ? `, ${webSocketClasses.join(', ')}` : ''} } from 'virtual:void-routes'; + return `${sandboxImport}${liveRuntimeImport}import app, { scheduled, __voidCollectPrerenderPaths, __voidHandleWebSocket, __withRawEnv${webSocketClasses.length > 0 ? `, ${webSocketClasses.join(', ')}` : ''}${liveRouteImport} } from 'virtual:void-routes'; ${runtimeEnvImport}${envSchemaImport}${remoteImport}${filterBindingsHelper}${assetHelper} ${internalRouteHelper} ${scheduledHandler} ${webSocketClasses.map((name) => `export { ${name} };`).join('\n')} +${liveExport} ${sandboxExport} export default { fetch: ${fetchHandler}, scheduled: handleScheduled };`; } diff --git a/packages/void/src/runtime/live-client.ts b/packages/void/src/runtime/live-client.ts new file mode 100644 index 000000000..570933afd --- /dev/null +++ b/packages/void/src/runtime/live-client.ts @@ -0,0 +1,427 @@ +import type { LiveClientEvent, LiveControlOperation, LiveControlResponse } from './live.ts'; + +export type LiveSubscribeOptions = { + id: string; + topic: string; + lastEventId?: string; + onEvent?: (event: LiveClientEvent) => void; +}; + +export type ConnectLiveStreamOptions = { + withCredentials?: boolean; + fetch?: typeof fetch; + headers?: HeadersInit | (() => HeadersInit | Promise); + eventSource?: EventSourceConstructor; + keepOpen?: boolean; + retryDelay?: number; + onError?: (error: Event | Error) => void; +}; + +export type LiveClient = { + readonly connectionId: string; + subscribe(options: LiveSubscribeOptions): Promise<() => Promise>; + close(): void; +}; + +type EventSourceLike = { + addEventListener(type: string, listener: (event: Event) => void): void; + removeEventListener(type: string, listener: (event: Event) => void): void; + close(): void; +}; + +type EventSourceConstructor = new ( + url: string, + options?: { withCredentials?: boolean }, +) => EventSourceLike; + +type Subscription = { + id: string; + topic: string; + lastEventId?: string; + onEvent?: (event: LiveClientEvent) => void; +}; + +type PendingOperation = { + operation: LiveControlOperation; + resolve: () => void; + reject: (error: unknown) => void; +}; + +export function connectLiveStream( + url: string | URL, + options: ConnectLiveStreamOptions = {}, +): LiveClient { + const EventSourceCtor = + options.eventSource ?? (globalThis as { EventSource?: EventSourceConstructor }).EventSource; + if (!EventSourceCtor) { + throw new Error('live: EventSource is not available in this runtime.'); + } + const ResolvedEventSource = EventSourceCtor; + const fetchImpl = options.fetch ?? fetch; + const connectionId = crypto.randomUUID(); + const subscriptions = new Map(); + const subscriptionVersions = new Map(); + let source: EventSourceLike | null = null; + let openPromise: Promise | null = null; + let rejectOpen: ((error: Error) => void) | null = null; + let reconnectTimer: ReturnType | null = null; + let resubscribeOnNextOpen = false; + let resubscribeWithNextFlush = false; + let closed = false; + let pendingOperations: Array = []; + let pendingFlush: Promise | null = null; + let flushChain: Promise = Promise.resolve(); + let nextSubscriptionVersion = 0; + const retryDelay = options.retryDelay ?? 1_000; + if (!Number.isFinite(retryDelay) || retryDelay < 0) { + throw new Error('live: retryDelay must be a non-negative finite number.'); + } + + function streamUrl(): string { + const resolved = new URL(String(url), globalThis.location?.href); + resolved.searchParams.set('connectionId', connectionId); + return resolved.href; + } + + function ensureOpen(): Promise { + if (closed) { + return Promise.reject(new Error('live: stream is closed.')); + } + const shouldResubscribe = resubscribeOnNextOpen; + if (reconnectTimer != null) { + clearTimeout(reconnectTimer); + reconnectTimer = null; + } + if (openPromise) { + return openPromise; + } + resubscribeOnNextOpen = false; + if (shouldResubscribe) { + resubscribeWithNextFlush = true; + } + return openSource(false); + } + + function openSource(resubscribeOnOpen: boolean): Promise { + const current = new ResolvedEventSource(streamUrl(), { + withCredentials: options.withCredentials, + }); + source = current; + let opened = false; + openPromise = new Promise((resolve, reject) => { + rejectOpen = reject; + const onOpen = (): void => { + if (source !== current) { + return; + } + if (!opened) { + opened = true; + rejectOpen = null; + resolve(); + if (resubscribeOnOpen) { + void resubscribeAll().catch(reportError); + } + return; + } + void resubscribeAll().catch(reportError); + }; + const onError = (event: Event): void => { + reportError(event); + if (source !== current) { + return; + } + if (!opened) { + current.removeEventListener('open', onOpen); + current.removeEventListener('error', onError); + current.removeEventListener('message', onMessage); + current.close(); + source = null; + openPromise = null; + rejectOpen = null; + reject(new Error('live: stream failed to open.')); + if (resubscribeOnOpen) { + scheduleReconnectTimer(); + } + return; + } + scheduleReconnect(current); + }; + current.addEventListener('open', onOpen); + current.addEventListener('error', onError); + current.addEventListener('message', onMessage); + }); + return openPromise; + } + + function scheduleReconnect(current: EventSourceLike): void { + if (closed || reconnectTimer != null) { + return; + } + current.close(); + if (source === current) { + source = null; + openPromise = null; + } + if (!options.keepOpen && subscriptions.size === 0) { + return; + } + scheduleReconnectTimer(); + } + + function scheduleReconnectTimer(): void { + if (closed || reconnectTimer != null || (!options.keepOpen && subscriptions.size === 0)) { + return; + } + resubscribeOnNextOpen = true; + reconnectTimer = setTimeout(() => { + reconnectTimer = null; + if (closed || (!options.keepOpen && subscriptions.size === 0)) { + resubscribeOnNextOpen = false; + return; + } + resubscribeOnNextOpen = false; + void openSource(true).catch((error) => { + if (!closed) { + reportError(error); + } + }); + }, retryDelay); + } + + function reportError(error: Event | Error | unknown): void { + if (error instanceof Event || error instanceof Error) { + options.onError?.(error); + } else { + options.onError?.(new Error(String(error))); + } + } + + function onMessage(raw: Event): void { + const message = raw as MessageEvent; + let event: LiveClientEvent; + try { + event = JSON.parse(String(message.data)) as LiveClientEvent; + } catch (error) { + reportError(error); + return; + } + const subscription = subscriptions.get(event.subscriptionId); + if (!subscription) { + return; + } + if (subscription.topic !== event.topic) { + return; + } + if (event.eventId) { + subscription.lastEventId = event.eventId; + } + subscription.onEvent?.(event); + } + + function enqueue(operation: LiveControlOperation): Promise { + const promise = new Promise((resolve, reject) => { + pendingOperations.push({ operation, resolve, reject }); + }); + scheduleFlush(); + return promise; + } + + function scheduleFlush(): void { + if (pendingFlush) { + return; + } + const scheduled = flushChain.then(flush); + pendingFlush = scheduled; + flushChain = scheduled.catch(() => {}); + void scheduled.finally(() => { + if (pendingFlush === scheduled) { + pendingFlush = null; + } + if (pendingOperations.length > 0) { + scheduleFlush(); + } + }); + } + + async function flush(): Promise { + const pending = pendingOperations; + pendingOperations = []; + if (pending.length === 0) { + return; + } + try { + await ensureOpen(); + let operations = pending.map((entry) => entry.operation); + if (resubscribeWithNextFlush) { + resubscribeWithNextFlush = false; + const pendingIds = new Set(operations.map((operation) => operation.id)); + operations = [ + ...[...subscriptions.values()] + .filter((subscription) => !pendingIds.has(subscription.id)) + .map( + (subscription): LiveControlOperation => ({ + kind: 'subscribe', + id: subscription.id, + topic: subscription.topic, + ...(subscription.lastEventId !== undefined && { + lastEventId: subscription.lastEventId, + }), + }), + ), + ...operations, + ]; + } + const headers = new Headers( + typeof options.headers === 'function' ? await options.headers() : options.headers, + ); + headers.set('content-type', 'application/json'); + const response = await fetchImpl(String(url), { + method: 'POST', + headers, + credentials: options.withCredentials ? 'include' : 'same-origin', + body: JSON.stringify({ + connectionId, + operations, + }), + }); + const result = (await response.json()) as LiveControlResponse; + if (!response.ok || !result.accepted) { + throw new Error(result.accepted ? response.statusText : result.error.message); + } + const remaining = [...result.results]; + for (const entry of pending) { + const index = remaining.findIndex( + (item) => item.id === entry.operation.id && item.kind === entry.operation.kind, + ); + const item = index === -1 ? undefined : remaining.splice(index, 1)[0]; + if (!item) { + entry.reject(new Error(`live: missing control result for ${entry.operation.id}.`)); + } else if (item.ok) { + entry.resolve(); + } else { + entry.reject(new Error(item.error.message)); + } + } + } catch (error) { + for (const entry of pending) { + entry.reject(error); + } + } + } + + async function resubscribeAll(): Promise { + const activeSubscriptions = [...subscriptions.values()]; + if (activeSubscriptions.length === 0) { + return; + } + await Promise.all( + activeSubscriptions.map((subscription) => + enqueue({ + kind: 'subscribe', + id: subscription.id, + topic: subscription.topic, + ...(subscription.lastEventId !== undefined && { lastEventId: subscription.lastEventId }), + }), + ), + ); + } + + function closeIfIdle(): void { + if (!options.keepOpen && subscriptions.size === 0) { + if (reconnectTimer != null) { + clearTimeout(reconnectTimer); + reconnectTimer = null; + } + resubscribeOnNextOpen = false; + resubscribeWithNextFlush = false; + source?.close(); + source = null; + openPromise = null; + rejectOpen = null; + } + } + + return { + connectionId, + async subscribe(subscribeOptions) { + const version = ++nextSubscriptionVersion; + const previousSubscription = subscriptions.get(subscribeOptions.id); + const previousVersion = subscriptionVersions.get(subscribeOptions.id); + subscriptionVersions.set(subscribeOptions.id, version); + const subscription: Subscription = { + id: subscribeOptions.id, + topic: subscribeOptions.topic, + lastEventId: subscribeOptions.lastEventId, + onEvent: subscribeOptions.onEvent as ((event: LiveClientEvent) => void) | undefined, + }; + subscriptions.set(subscription.id, subscription); + try { + await enqueue({ + kind: 'subscribe', + id: subscription.id, + topic: subscription.topic, + ...(subscription.lastEventId !== undefined && { lastEventId: subscription.lastEventId }), + }); + } catch (error) { + if (subscriptionVersions.get(subscription.id) === version) { + if (previousSubscription) { + subscriptions.set(subscription.id, previousSubscription); + if (previousVersion != null) { + subscriptionVersions.set(subscription.id, previousVersion); + } else { + subscriptionVersions.delete(subscription.id); + } + } else { + subscriptions.delete(subscription.id); + subscriptionVersions.delete(subscription.id); + } + } + queueMicrotask(closeIfIdle); + throw error; + } + if (closed) { + throw new Error('live: stream is closed.'); + } + let unsubscribed = false; + return async () => { + if (unsubscribed) { + return; + } + unsubscribed = true; + if ( + subscriptionVersions.get(subscription.id) !== version || + subscriptions.get(subscription.id) !== subscription + ) { + return; + } + await enqueue({ kind: 'unsubscribe', id: subscription.id }); + if (subscriptions.get(subscription.id) === subscription) { + subscriptions.delete(subscription.id); + subscriptionVersions.delete(subscription.id); + } + closeIfIdle(); + }; + }, + close() { + const closeError = new Error('live: stream is closed.'); + closed = true; + subscriptions.clear(); + subscriptionVersions.clear(); + if (reconnectTimer != null) { + clearTimeout(reconnectTimer); + reconnectTimer = null; + } + resubscribeOnNextOpen = false; + resubscribeWithNextFlush = false; + for (const entry of pendingOperations) { + entry.reject(closeError); + } + pendingOperations = []; + rejectOpen?.(closeError); + rejectOpen = null; + source?.close(); + source = null; + openPromise = null; + }, + }; +} diff --git a/packages/void/src/runtime/live-server.ts b/packages/void/src/runtime/live-server.ts new file mode 100644 index 000000000..9bd2e9bba --- /dev/null +++ b/packages/void/src/runtime/live-server.ts @@ -0,0 +1,650 @@ +import { + __getLiveBindingName, + type LiveClientEvent, + type LiveControlErrorCode, + type LiveControlOperation, + type LiveControlOperationResult, + type LiveControlResponse, + type LiveLimits, +} from './live.ts'; +import { eventStream, type SseStream } from './sse.ts'; + +type LiveDurableObjectEnv = Record; + +type ConnectRequest = { + streamId: string; + connectionId: string; + owner: string | null; + limits: LiveLimits; +}; + +type ControlRequest = ConnectRequest & { + operations: Array; +}; + +type PublishRequest = { + streamId: string; + topic: string; + topicKey: string; + data: unknown; + type?: string; + eventId?: string; + limits: LiveLimits; +}; + +type DeliveryRequest = { + streamId: string; + topic: string; + data: unknown; + type?: string; + eventId?: string; + subscriptions: Array; +}; + +type DeliveryResponse = { + delivered: number; + stale: Array; +}; + +type StaleCheckRequest = { + subscriptions: Array; +}; + +type StaleCheckResponse = { + stale: Array; +}; + +type SubscriberRow = { + streamId: string; + topic: string; + topicKey: string; + connectionId: string; + subscriptionId: string; + generation: number; + revision: number; + lastSeenAt: number; +}; + +type ActiveSubscription = { + id: string; + topic: string; + topicKey: string; + generation: number; + revision: number; + active: boolean; +}; + +type ActiveConnection = { + stream: SseStream; + owner: string | null; + generation: number; + subscriptions: Map; + writeChain: Promise; + queued: number; + limits: LiveLimits; +}; + +export interface LiveDurableObjectInstance { + fetch(request: Request): Promise; + alarm?(): Promise; +} + +export interface LiveDurableObjectClass { + new (state: DurableObjectState, env: LiveDurableObjectEnv): LiveDurableObjectInstance; +} + +const GENERATION_KEY = 'connection:generation'; +const TOPIC_PRUNE_ALARM_DELAY_MS = 60_000; +const textEncoder = new TextEncoder(); + +export function createLiveDurableObject(): LiveDurableObjectClass { + class VoidLiveStreamDurableObject implements LiveDurableObjectInstance { + private connection: ActiveConnection | null = null; + private publishChain: Promise = Promise.resolve(); + + constructor( + public readonly state: DurableObjectState, + public readonly env: LiveDurableObjectEnv, + ) {} + + async fetch(request: Request): Promise { + const { pathname } = new URL(request.url); + if (request.method !== 'POST') { + return new Response('Method Not Allowed', { status: 405 }); + } + if (pathname === '/connect') { + return this.connect((await request.json()) as ConnectRequest); + } + if (pathname === '/control') { + return this.control((await request.json()) as ControlRequest); + } + if (pathname === '/deliver') { + return this.deliver((await request.json()) as DeliveryRequest); + } + if (pathname === '/check') { + return this.check((await request.json()) as StaleCheckRequest); + } + if (pathname === '/register') { + return this.register((await request.json()) as SubscriberRow & { limits: LiveLimits }); + } + if (pathname === '/unregister') { + return this.unregister((await request.json()) as Partial); + } + if (pathname === '/publish') { + return this.enqueuePublish(request); + } + return new Response('Not Found', { status: 404 }); + } + + async alarm(): Promise { + await this.pruneAllSubscribers(); + } + + private enqueuePublish(request: Request): Promise { + const run = this.publishChain + .catch(() => {}) + .then(async () => { + return this.publish((await request.json()) as PublishRequest); + }); + this.publishChain = run.then( + () => {}, + () => {}, + ); + return run; + } + + private async connect(message: ConnectRequest): Promise { + const previous = this.connection; + if (previous) { + const active = [...previous.subscriptions.values()].filter((sub) => sub.active); + for (const subscription of previous.subscriptions.values()) { + subscription.active = false; + } + await previous.stream.close().catch(() => {}); + await Promise.all( + active.map((sub) => this.unregisterTopic(message.streamId, message.connectionId, sub)), + ); + } + + const generation = ((await this.state.storage.get(GENERATION_KEY)) ?? 0) + 1; + await this.state.storage.put(GENERATION_KEY, generation); + + return eventStream( + async (stream) => { + const subscriptions = + previous?.owner === message.owner ? previous.subscriptions : new Map(); + for (const subscription of subscriptions.values()) { + subscription.active = false; + } + this.connection = { + stream, + owner: message.owner, + generation, + subscriptions, + writeChain: Promise.resolve(), + queued: 0, + limits: message.limits, + }; + await stream.comment('connected'); + await stream.closed; + if (this.connection?.generation === generation) { + const active = [...this.connection.subscriptions.values()].filter((sub) => sub.active); + this.connection = null; + await Promise.all( + active.map((sub) => + this.unregisterTopic(message.streamId, message.connectionId, sub), + ), + ); + } + }, + { keepAlive: { intervalMs: 15_000, comment: 'keep-alive' } }, + ); + } + + private async control(message: ControlRequest): Promise { + const connection = this.connection; + if (!connection) { + return controlBatchError('CONNECTION_NOT_FOUND', 'live: connection is not open.', 404); + } + if (connection.owner !== message.owner) { + return controlBatchError('OWNER_MISMATCH', 'live: connection owner mismatch.', 403); + } + + const results: Array = []; + for (const operation of message.operations) { + if (operation.kind === 'unsubscribe') { + const existing = connection.subscriptions.get(operation.id); + if (existing) { + existing.active = false; + await this.unregisterTopic(message.streamId, message.connectionId, existing); + connection.subscriptions.delete(operation.id); + } + results.push({ ok: true, kind: 'unsubscribe', id: operation.id }); + continue; + } + + if ( + !connection.subscriptions.has(operation.id) && + activeSubscriptionCount(connection) >= message.limits.maxSubscriptionsPerConnection + ) { + results.push({ + ok: false, + kind: 'subscribe', + id: operation.id, + error: { code: 'SUBSCRIPTION_LIMIT', message: 'live: subscription limit reached.' }, + }); + continue; + } + + const topicKey = await topicDigest(message.streamId, operation.topic); + const existing = connection.subscriptions.get(operation.id); + const revision = (existing?.revision ?? 0) + 1; + const row: SubscriberRow = { + streamId: message.streamId, + topic: operation.topic, + topicKey, + connectionId: message.connectionId, + subscriptionId: operation.id, + generation: connection.generation, + revision, + lastSeenAt: Date.now(), + }; + const response = await this.topicStub(message.streamId, topicKey).fetch( + internalUrl('/register'), + { + method: 'POST', + body: JSON.stringify({ ...row, limits: message.limits }), + }, + ); + if (!response.ok) { + results.push({ + ok: false, + kind: 'subscribe', + id: operation.id, + error: + response.status === 409 + ? { code: 'TOPIC_FULL', message: 'live: topic subscription limit reached.' } + : { code: 'INTERNAL_ERROR', message: await response.text() }, + }); + continue; + } + + if (existing) { + existing.active = false; + await this.unregisterTopic(message.streamId, message.connectionId, existing).catch( + () => {}, + ); + } + connection.subscriptions.set(operation.id, { + id: operation.id, + topic: operation.topic, + topicKey, + generation: connection.generation, + revision, + active: true, + }); + results.push({ ok: true, kind: 'subscribe', id: operation.id, topic: operation.topic }); + } + + return Response.json({ + accepted: true, + connectionId: message.connectionId, + results, + } satisfies LiveControlResponse); + } + + private async deliver(message: DeliveryRequest): Promise { + const connection = this.connection; + if (!connection) { + return new Response('connection missing', { status: 410 }); + } + + let delivered = 0; + const stale: Array = []; + for (const [index, row] of message.subscriptions.entries()) { + if (isStaleSubscriber(row, connection)) { + stale.push(index); + continue; + } + if (connection.queued >= connection.limits.maxQueuedEventsPerConnection) { + await connection.stream.close().catch(() => {}); + this.connection = null; + return new Response('connection queue full', { status: 410 }); + } + const event: LiveClientEvent = { + subscriptionId: row.subscriptionId, + topic: message.topic, + ...(message.type !== undefined && { type: message.type }), + ...(message.eventId !== undefined && { eventId: message.eventId }), + data: message.data, + }; + if (encodedLiveEventSize(event) > connection.limits.maxEncodedEventSize) { + return payloadTooLargeResponse(connection.limits.maxEncodedEventSize); + } + connection.queued++; + const activeConnection = connection; + connection.writeChain = connection.writeChain + .catch(() => {}) + .then(() => connection.stream.send({ event: 'message', data: event })) + .catch(() => { + if (this.connection === activeConnection) { + this.connection = null; + } + void connection.stream.close().catch(() => {}); + }) + .finally(() => { + connection.queued--; + }); + delivered++; + } + return Response.json({ delivered, stale } satisfies DeliveryResponse); + } + + private check(message: StaleCheckRequest): Response { + const connection = this.connection; + if (!connection) { + return new Response('connection missing', { status: 410 }); + } + + const stale: Array = []; + for (const [index, row] of message.subscriptions.entries()) { + if (isStaleSubscriber(row, connection)) { + stale.push(index); + } + } + return Response.json({ stale } satisfies StaleCheckResponse); + } + + private async register(row: SubscriberRow & { limits: LiveLimits }): Promise { + const prefix = subscriberPrefix(row.streamId, row.topicKey); + const current = await this.state.storage.list({ prefix }); + const deletedKeys = new Set(); + for (const [key, value] of current) { + if (value.connectionId === row.connectionId && value.generation < row.generation) { + deletedKeys.add(key); + continue; + } + if ( + value.connectionId === row.connectionId && + value.subscriptionId === row.subscriptionId + ) { + deletedKeys.add(key); + } + } + + let activeRows = [...current].filter(([key]) => !deletedKeys.has(key)); + if (activeRows.length >= row.limits.maxSubscriptionsPerTopic) { + for (const key of await this.findStaleSubscriberKeys(activeRows, row.limits)) { + deletedKeys.add(key); + } + activeRows = activeRows.filter(([key]) => !deletedKeys.has(key)); + } + + if (activeRows.length >= row.limits.maxSubscriptionsPerTopic) { + return new Response('topic full', { status: 409 }); + } + if (deletedKeys.size > 0) { + await this.state.storage.delete([...deletedKeys]); + } + await this.state.storage.put(subscriberKey(row), withoutLimits(row)); + await this.schedulePruneAlarm(); + return Response.json({ ok: true }); + } + + private async unregister(row: Partial): Promise { + if (!row.streamId || !row.topicKey) { + return Response.json({ ok: true }); + } + const key = + row.connectionId && row.subscriptionId && row.generation != null && row.revision != null + ? subscriberKey(row as SubscriberRow) + : undefined; + if (key) { + await this.state.storage.delete(key); + } + return Response.json({ ok: true }); + } + + private async publish(message: PublishRequest): Promise { + const rows = await this.state.storage.list({ + prefix: subscriberPrefix(message.streamId, message.topicKey), + }); + const grouped = new Map>(); + for (const [key, row] of rows) { + const list = grouped.get(row.connectionId) ?? []; + list.push({ key, row }); + grouped.set(row.connectionId, list); + } + for (const { row } of [...grouped.values()].flat()) { + const event: LiveClientEvent = { + subscriptionId: row.subscriptionId, + topic: message.topic, + ...(message.type !== undefined && { type: message.type }), + ...(message.eventId !== undefined && { eventId: message.eventId }), + data: message.data, + }; + if (encodedLiveEventSize(event) > message.limits.maxEncodedEventSize) { + return payloadTooLargeResponse(message.limits.maxEncodedEventSize); + } + } + for (const [connectionId, items] of grouped) { + const response = await fetchWithTimeout( + this.connectionStub(message.streamId, connectionId), + internalUrl('/deliver'), + { + method: 'POST', + body: JSON.stringify({ + streamId: message.streamId, + topic: message.topic, + data: message.data, + type: message.type, + eventId: message.eventId, + subscriptions: items.map((item) => item.row), + } satisfies DeliveryRequest), + }, + message.limits.deliveryAttemptTimeoutMs, + ); + if (!response || response.status === 410 || response.status === 404) { + await this.state.storage.delete(items.map((item) => item.key)); + continue; + } + if (response.status === 413) { + return response; + } + if (response.ok) { + const result = (await response.json().catch(() => null)) as DeliveryResponse | null; + const staleKeys = result?.stale + .map((index) => items[index]?.key) + .filter((key): key is string => typeof key === 'string'); + if (staleKeys?.length) { + await this.state.storage.delete(staleKeys); + } + } + } + return Response.json({ ok: true }); + } + + private async pruneAllSubscribers(): Promise { + const rows = await this.state.storage.list({ prefix: 'sub:' }); + const staleKeys = await this.findStaleSubscriberKeys([...rows], { + deliveryAttemptTimeoutMs: 1_500, + }); + if (staleKeys.length > 0) { + await this.state.storage.delete(staleKeys); + } + if (rows.size > staleKeys.length) { + await this.schedulePruneAlarm(); + } + } + + private async findStaleSubscriberKeys( + entries: Array<[string, SubscriberRow]>, + limits: Pick, + ): Promise> { + const grouped = new Map>(); + for (const [key, row] of entries) { + const list = grouped.get(row.connectionId) ?? []; + list.push({ key, row }); + grouped.set(row.connectionId, list); + } + + const staleKeys: Array = []; + for (const [connectionId, items] of grouped) { + const response = await fetchWithTimeout( + this.connectionStub(items[0]!.row.streamId, connectionId), + internalUrl('/check'), + { + method: 'POST', + body: JSON.stringify({ + subscriptions: items.map((item) => item.row), + } satisfies StaleCheckRequest), + }, + limits.deliveryAttemptTimeoutMs, + ); + if (!response || response.status === 410 || response.status === 404) { + staleKeys.push(...items.map((item) => item.key)); + continue; + } + if (response.ok) { + const result = (await response.json().catch(() => null)) as StaleCheckResponse | null; + for (const index of result?.stale ?? []) { + const key = items[index]?.key; + if (key) { + staleKeys.push(key); + } + } + } + } + return staleKeys; + } + + private async schedulePruneAlarm(): Promise { + await this.state.storage.setAlarm?.(Date.now() + TOPIC_PRUNE_ALARM_DELAY_MS); + } + + private topicStub(streamId: string, topicKey: string): DurableObjectStub { + const binding = this.env[__getLiveBindingName()] as DurableObjectNamespace; + return binding.get(binding.idFromName(topicInstanceName(streamId, topicKey))); + } + + private connectionStub(streamId: string, connectionId: string): DurableObjectStub { + const binding = this.env[__getLiveBindingName()] as DurableObjectNamespace; + return binding.get(binding.idFromName(connectionInstanceName(streamId, connectionId))); + } + + private unregisterTopic( + streamId: string, + connectionId: string, + subscription: ActiveSubscription, + ) { + return this.topicStub(streamId, subscription.topicKey).fetch(internalUrl('/unregister'), { + method: 'POST', + body: JSON.stringify({ + streamId, + topic: subscription.topic, + topicKey: subscription.topicKey, + connectionId, + subscriptionId: subscription.id, + revision: subscription.revision, + generation: subscription.generation, + }), + }); + } + } + + return VoidLiveStreamDurableObject; +} + +function controlBatchError(code: LiveControlErrorCode, message: string, status: number): Response { + return Response.json({ accepted: false, error: { code, message } }, { status }); +} + +function activeSubscriptionCount(connection: ActiveConnection): number { + let count = 0; + for (const subscription of connection.subscriptions.values()) { + if (subscription.active) { + count++; + } + } + return count; +} + +function isStaleSubscriber(row: SubscriberRow, connection: ActiveConnection): boolean { + if (row.generation !== connection.generation) { + return true; + } + const subscription = connection.subscriptions.get(row.subscriptionId); + return !subscription || !subscription.active || subscription.revision !== row.revision; +} + +function encodedLiveEventSize(event: LiveClientEvent): number { + return textEncoder.encode(JSON.stringify(event)).byteLength; +} + +function payloadTooLargeResponse(maxEncodedEventSize: number): Response { + return Response.json( + { + accepted: false, + error: { + code: 'PAYLOAD_TOO_LARGE', + message: `live: encoded event exceeds ${maxEncodedEventSize} bytes.`, + }, + } satisfies LiveControlResponse, + { status: 413 }, + ); +} + +async function fetchWithTimeout( + stub: DurableObjectStub, + input: RequestInfo | URL, + init: RequestInit, + timeoutMs: number, +): Promise { + const fetchPromise = stub.fetch(input, init); + fetchPromise.catch(() => {}); + let timeout: ReturnType | undefined; + const timeoutPromise = new Promise((resolve) => { + timeout = setTimeout(() => resolve(null), Math.max(1, timeoutMs)); + }); + try { + return await Promise.race([fetchPromise, timeoutPromise]); + } catch { + return null; + } finally { + if (timeout) { + clearTimeout(timeout); + } + } +} + +function withoutLimits(row: SubscriberRow & { limits?: LiveLimits }): SubscriberRow { + const { limits: _limits, ...rest } = row; + return rest; +} + +function subscriberPrefix(streamId: string, topicKey: string): string { + return `sub:${streamId}:${topicKey}:`; +} + +function subscriberKey(row: SubscriberRow): string { + return `${subscriberPrefix(row.streamId, row.topicKey)}${row.connectionId}:${row.subscriptionId}:${row.generation}:${row.revision}`; +} + +async function topicDigest(streamId: string, topic: string): Promise { + const bytes = new TextEncoder().encode(`${streamId}\u0000${topic}`); + const digest = await crypto.subtle.digest('SHA-256', bytes); + const binary = String.fromCharCode(...new Uint8Array(digest)); + return btoa(binary).replaceAll('+', '-').replaceAll('/', '_').replace(/=+$/, ''); +} + +function connectionInstanceName(streamId: string, connectionId: string): string { + return `live:${streamId}:connection:${connectionId}`; +} + +function topicInstanceName(streamId: string, topicKey: string): string { + return `live:${streamId}:topic:${topicKey}`; +} + +function internalUrl(path: string): string { + return `https://void.live${path}`; +} diff --git a/packages/void/src/runtime/live.ts b/packages/void/src/runtime/live.ts new file mode 100644 index 000000000..37f9bf1c1 --- /dev/null +++ b/packages/void/src/runtime/live.ts @@ -0,0 +1,658 @@ +import { getUser, type AuthUser } from './auth.ts'; +import { getRawRuntimeEnv } from './env-raw.ts'; +import type { CloudContext, CloudEnv } from './handler.ts'; + +type MaybePromise = T | Promise; + +export type LiveControlRequest = { + connectionId: string; + operations: Array; +}; + +type ParsedControlOperation = + | { + ok: true; + index: number; + operation: LiveControlOperation; + } + | { + ok: false; + index: number; + result: LiveControlOperationResult; + }; + +type ParsedLiveControlRequest = { + connectionId: string; + operations: Array; +}; + +export type LiveControlOperation = + | { + kind: 'subscribe'; + id: string; + topic: string; + lastEventId?: string; + } + | { + kind: 'unsubscribe'; + id: string; + }; + +export type LiveControlErrorCode = + | 'BAD_REQUEST' + | 'CONNECTION_CLOSED' + | 'CONNECTION_NOT_FOUND' + | 'FORBIDDEN' + | 'INVALID_SUBSCRIPTION_ID' + | 'INVALID_TOPIC' + | 'OWNER_MISMATCH' + | 'SUBSCRIPTION_LIMIT' + | 'TOPIC_FULL' + | 'PAYLOAD_TOO_LARGE' + | 'INTERNAL_ERROR'; + +export type LiveControlError = { + code: LiveControlErrorCode; + message: string; +}; + +export type LiveControlOperationResult = + | { + ok: true; + kind: 'subscribe' | 'unsubscribe'; + id: string; + topic?: string; + } + | { + ok: false; + kind: 'subscribe' | 'unsubscribe'; + id: string; + error: LiveControlError; + }; + +export type LiveControlResponse = + | { + accepted: true; + connectionId: string; + results: Array; + } + | { + accepted: false; + error: LiveControlError; + }; + +export type LiveClientEvent = { + subscriptionId: string; + topic: string; + type?: string; + eventId?: string; + data: Data; +}; + +export type LivePublishOptions = { + type?: string; + eventId?: string; +}; + +export type LiveConnectOptions = { + owner?: string | null; + user?: unknown; +}; + +export type LiveControlOptions = LiveConnectOptions; + +export type LiveHookContext = { + request: Request; + env: CloudEnv['Bindings']; + user: AuthUser | unknown | null; + owner: string | null; +}; + +export type LiveConnectHookContext = LiveHookContext & { + connection: { id: string }; +}; + +export type LiveSubscribeHookContext = LiveHookContext & { + connection: { id: string }; + subscription: { id: string }; + topic: string; + lastEventId?: string; +}; + +export type LiveStreamOptions = { + id: string; + allowAnonymousControl?: boolean; + identifyConnection?: (ctx: LiveHookContext) => MaybePromise; + onConnect?: (ctx: LiveConnectHookContext) => MaybePromise; + onSubscribe?: (ctx: LiveSubscribeHookContext) => MaybePromise; + limits?: Partial; +}; + +export type LiveLimits = { + maxSubscriptionsPerConnection: number; + maxSubscriptionsPerTopic: number; + maxOperationsPerControlRequest: number; + maxQueuedEventsPerConnection: number; + maxEncodedEventSize: number; + deliveryAttemptTimeoutMs: number; +}; + +export type LiveStream = { + connect(ctx: CloudContext, options?: LiveConnectOptions): Promise; + control(ctx: CloudContext, options?: LiveControlOptions): Promise; + publish(topic: string, data: unknown, options?: LivePublishOptions): Promise; + withEnv(env: unknown): Pick; +}; + +let liveBindingName = 'VOID_LIVE'; +const registeredStreamIds = new Set(); +const DEFAULT_LIMITS: LiveLimits = { + maxSubscriptionsPerConnection: 256, + maxSubscriptionsPerTopic: 256, + maxOperationsPerControlRequest: 100, + maxQueuedEventsPerConnection: 100, + maxEncodedEventSize: 64 * 1024, + deliveryAttemptTimeoutMs: 1_500, +}; +const MAX_SUBSCRIPTIONS_PER_TOPIC = 256; + +export function __setLiveBindingName(name: string): void { + liveBindingName = assertBindingName(name); +} + +export function __getLiveBindingName(): string { + return liveBindingName; +} + +export function defineLiveStream(options: LiveStreamOptions): LiveStream { + const streamId = assertStreamId(options.id); + const limits = resolveLiveLimits(options.limits); + if (registeredStreamIds.has(streamId)) { + throw new Error(`live: duplicate stream id "${streamId}".`); + } + registeredStreamIds.add(streamId); + + async function resolveOwner( + ctx: CloudContext, + request: Request, + connectOptions: LiveConnectOptions | undefined, + ): Promise<{ owner: string | null; user: unknown | null }> { + const hasExplicitUser = connectOptions && 'user' in connectOptions; + const resolvedUser = hasExplicitUser ? connectOptions.user : (getContextUser(ctx) ?? getUser()); + const hookContext: LiveHookContext = { + request, + env: ctx.env, + user: resolvedUser ?? null, + owner: connectOptions?.owner ?? null, + }; + const identified = + connectOptions && 'owner' in connectOptions + ? connectOptions.owner + : await options.identifyConnection?.(hookContext); + const owner = identified ?? defaultOwner(resolvedUser); + if (owner == null && !options.allowAnonymousControl) { + throw liveResponseError('FORBIDDEN', 'live: anonymous control is not enabled.', 403); + } + assertOwner(owner); + return { owner, user: resolvedUser ?? null }; + } + + async function publishWithEnv( + env: unknown, + topic: string, + data: unknown, + publishOptions: LivePublishOptions = {}, + ): Promise { + const binding = getLiveBinding(env); + assertTopic(topic); + assertOptionalString('type', publishOptions.type); + assertOptionalString('eventId', publishOptions.eventId); + assertJsonSerializable(data); + const topicKey = await topicDigest(streamId, topic); + const id = binding.idFromName(topicInstanceName(streamId, topicKey)); + const response = await binding.get(id).fetch(internalUrl('/publish'), { + method: 'POST', + body: JSON.stringify({ + streamId, + topic, + topicKey, + data, + type: publishOptions.type, + eventId: publishOptions.eventId, + limits, + }), + }); + if (!response.ok) { + const body = await response.text(); + const parsed = parseLiveControlResponse(body); + if (parsed && !parsed.accepted) { + throw new Error(`${parsed.error.code}: ${parsed.error.message}`); + } + throw new Error(body || response.statusText); + } + } + + return { + async connect(ctx, connectOptions) { + try { + const request = ctx.req.raw; + const connectionId = getConnectionId(request); + const { owner, user } = await resolveOwner(ctx, request, connectOptions); + const decision = await options.onConnect?.({ + request, + env: ctx.env, + user, + owner, + connection: { id: connectionId }, + }); + if (decision instanceof Response) { + return decision; + } + const binding = getLiveBinding(ctx.env); + const id = binding.idFromName(connectionInstanceName(streamId, connectionId)); + return binding.get(id).fetch(internalUrl('/connect'), { + method: 'POST', + body: JSON.stringify({ streamId, connectionId, owner, limits }), + }); + } catch (error) { + return errorToResponse(error); + } + }, + + async control(ctx, controlOptions) { + try { + const request = ctx.req.raw; + const body = await readControlRequest(request, limits); + const { owner, user } = await resolveOwner(ctx, request, controlOptions); + const authorized: Array<{ operation: LiveControlOperation; index: number }> = []; + const results: Array = []; + for (const entry of body.operations) { + if (!entry.ok) { + results[entry.index] = entry.result; + continue; + } + const { operation, index } = entry; + if (operation.kind === 'unsubscribe') { + authorized.push({ operation, index }); + continue; + } + const decision = await options.onSubscribe?.({ + request, + env: ctx.env, + user, + owner, + connection: { id: body.connectionId }, + subscription: { id: operation.id }, + topic: operation.topic, + lastEventId: operation.lastEventId, + }); + if (decision instanceof Response) { + results[index] = { + ok: false, + kind: 'subscribe', + id: operation.id, + error: await responseToControlError(decision), + }; + continue; + } + authorized.push({ operation, index }); + } + if (authorized.length > 0) { + const binding = getLiveBinding(ctx.env); + const id = binding.idFromName(connectionInstanceName(streamId, body.connectionId)); + const response = await binding.get(id).fetch(internalUrl('/control'), { + method: 'POST', + body: JSON.stringify({ + streamId, + connectionId: body.connectionId, + owner, + operations: authorized.map((entry) => entry.operation), + limits, + }), + }); + const controlResponse = (await response.json()) as LiveControlResponse; + if (!controlResponse.accepted) { + return Response.json(controlResponse, { status: response.status }); + } + for (const [resultIndex, result] of controlResponse.results.entries()) { + const operation = authorized[resultIndex]; + if (operation) { + results[operation.index] = result; + } + } + } + for (const entry of body.operations) { + if (!entry.ok) { + continue; + } + const { operation, index } = entry; + results[index] ??= { + ok: false, + kind: operation.kind, + id: operation.id, + error: { code: 'INTERNAL_ERROR', message: 'live: missing control result.' }, + }; + } + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: results as Array, + } satisfies LiveControlResponse); + } catch (error) { + return errorToResponse(error); + } + }, + + publish(topic, data, publishOptions) { + let env: unknown; + try { + env = getRawRuntimeEnv(); + } catch { + throw new Error( + 'live: publish() was called outside a Void server runtime context. Pass an explicit env with stream.withEnv(env).publish(...).', + ); + } + return publishWithEnv(env, topic, data, publishOptions); + }, + + withEnv(env) { + return { + publish(topic, data, publishOptions) { + return publishWithEnv(env, topic, data, publishOptions); + }, + }; + }, + }; +} + +function defaultOwner(user: unknown): string | null { + if (user && typeof user === 'object' && 'id' in user && typeof user.id === 'string') { + return `user:${user.id}`; + } + return null; +} + +function getContextUser(ctx: CloudContext): unknown | null { + try { + return ctx.get('user'); + } catch { + return null; + } +} + +function parseLiveControlResponse(body: string): LiveControlResponse | null { + try { + const parsed = JSON.parse(body) as LiveControlResponse; + return typeof parsed === 'object' && parsed !== null && 'accepted' in parsed ? parsed : null; + } catch { + return null; + } +} + +async function readControlRequest( + request: Request, + limits: LiveLimits, +): Promise { + let body: unknown; + try { + body = await request.json(); + } catch { + throw liveResponseError('BAD_REQUEST', 'live: control request body must be JSON.', 400); + } + if (typeof body !== 'object' || body === null || Array.isArray(body)) { + throw liveResponseError('BAD_REQUEST', 'live: control request body must be an object.', 400); + } + const record = body as Record; + const connectionId = assertConnectionId(record.connectionId); + if (!Array.isArray(record.operations)) { + throw liveResponseError('BAD_REQUEST', 'live: operations must be an array.', 400); + } + if (record.operations.length > limits.maxOperationsPerControlRequest) { + throw liveResponseError( + 'BAD_REQUEST', + 'live: too many operations in one control request.', + 400, + ); + } + const operations = record.operations.map(parseOperation); + return { connectionId, operations }; +} + +function parseOperation(value: unknown, index: number): ParsedControlOperation { + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + throw liveResponseError('BAD_REQUEST', 'live: operation must be an object.', 400); + } + const record = value as Record; + if (record.kind === 'subscribe') { + const id = parseBoundedString(record.id, 1, 256); + if (id == null) { + return invalidOperation(index, 'subscribe', record.id, 'INVALID_SUBSCRIPTION_ID'); + } + const topic = parseBoundedString(record.topic, 1, 512); + if (topic == null) { + return invalidOperation(index, 'subscribe', id, 'INVALID_TOPIC'); + } + let lastEventId: string | undefined; + if (record.lastEventId !== undefined) { + lastEventId = parseBoundedString(record.lastEventId, 0, 512) ?? undefined; + if (lastEventId === undefined) { + return invalidOperation(index, 'subscribe', id, 'BAD_REQUEST'); + } + } + return { + ok: true, + index, + operation: { + kind: 'subscribe', + id, + topic, + ...(lastEventId !== undefined && { lastEventId }), + }, + }; + } + if (record.kind === 'unsubscribe') { + const id = parseBoundedString(record.id, 1, 256); + if (id == null) { + return invalidOperation(index, 'unsubscribe', record.id, 'INVALID_SUBSCRIPTION_ID'); + } + return { ok: true, index, operation: { kind: 'unsubscribe', id } }; + } + throw liveResponseError('BAD_REQUEST', 'live: operation kind is invalid.', 400); +} + +function invalidOperation( + index: number, + kind: 'subscribe' | 'unsubscribe', + id: unknown, + code: LiveControlErrorCode, +): ParsedControlOperation { + const messages: Record = { + BAD_REQUEST: 'live: invalid control operation.', + CONNECTION_CLOSED: 'live: connection closed.', + CONNECTION_NOT_FOUND: 'live: connection not found.', + FORBIDDEN: 'Forbidden', + INVALID_SUBSCRIPTION_ID: 'live: invalid subscription id.', + INVALID_TOPIC: 'live: invalid topic.', + OWNER_MISMATCH: 'live: owner mismatch.', + SUBSCRIPTION_LIMIT: 'live: subscription limit reached.', + TOPIC_FULL: 'live: topic subscription limit reached.', + PAYLOAD_TOO_LARGE: 'live: payload too large.', + INTERNAL_ERROR: 'live: internal error.', + }; + return { + ok: false, + index, + result: { + ok: false, + kind, + id: typeof id === 'string' ? id : '', + error: { code, message: messages[code] }, + }, + }; +} + +function getLiveBinding(env: unknown): DurableObjectNamespace { + if (typeof env !== 'object' || env === null) { + throw new Error('live: Cloudflare env is unavailable.'); + } + const binding = (env as Record)[liveBindingName] as + | DurableObjectNamespace + | undefined; + if (!binding) { + throw new Error(`live: Missing Durable Object binding "${liveBindingName}".`); + } + return binding; +} + +function getConnectionId(request: Request): string { + return assertConnectionId(new URL(request.url).searchParams.get('connectionId')); +} + +function assertStreamId(value: unknown): string { + return assertBoundedString('stream id', value, 1, 128); +} + +function assertBindingName(value: string): string { + if (!/^[A-Za-z_][A-Za-z0-9_]*$/.test(value)) { + throw new Error(`live: invalid Durable Object binding name "${value}".`); + } + return value; +} + +function assertConnectionId(value: unknown): string { + return assertBoundedString('connectionId', value, 16, 256); +} + +function assertTopic(value: unknown): string { + return assertBoundedString('topic', value, 1, 512); +} + +function assertOwner(value: string | null): void { + if (value !== null) { + assertBoundedString('owner', value, 1, 512); + } +} + +function assertOptionalString(name: string, value: unknown): void { + if (value !== undefined) { + assertBoundedString(name, value, 1, 512); + } +} + +function resolveLiveLimits(input: Partial | undefined): LiveLimits { + const limits = { ...DEFAULT_LIMITS }; + for (const key of Object.keys(limits) as Array) { + const value = input?.[key]; + if (value === undefined) { + continue; + } + if (!Number.isFinite(value) || !Number.isInteger(value) || value <= 0) { + throw new Error(`live: limits.${key} must be a positive finite integer.`); + } + limits[key] = value; + } + if (limits.maxSubscriptionsPerTopic > MAX_SUBSCRIPTIONS_PER_TOPIC) { + throw new Error( + `live: limits.maxSubscriptionsPerTopic must be at most ${MAX_SUBSCRIPTIONS_PER_TOPIC}.`, + ); + } + return limits; +} + +function assertBoundedString(name: string, value: unknown, min: number, max: number): string { + const parsed = parseBoundedString(value, min, max); + if (parsed == null) { + throw liveResponseError('BAD_REQUEST', `live: invalid ${name}.`, 400); + } + return parsed; +} + +function parseBoundedString(value: unknown, min: number, max: number): string | null { + if ( + typeof value !== 'string' || + value.length < min || + value.length > max || + /[\r\n]/.test(value) || + value.includes('\u0000') + ) { + return null; + } + return value; +} + +function assertJsonSerializable(value: unknown): void { + try { + const serialized = JSON.stringify(value); + if (typeof serialized !== 'string') { + throw new Error('unsupported value'); + } + } catch (error) { + throw new Error( + `live: payload must be JSON-serializable. ${error instanceof Error ? error.message : String(error)}`, + ); + } +} + +async function topicDigest(streamId: string, topic: string): Promise { + const bytes = new TextEncoder().encode(`${streamId}\u0000${topic}`); + const digest = await crypto.subtle.digest('SHA-256', bytes); + const binary = String.fromCharCode(...new Uint8Array(digest)); + return btoa(binary).replaceAll('+', '-').replaceAll('/', '_').replace(/=+$/, ''); +} + +function connectionInstanceName(streamId: string, connectionId: string): string { + return `live:${streamId}:connection:${connectionId}`; +} + +function topicInstanceName(streamId: string, topicKey: string): string { + return `live:${streamId}:topic:${topicKey}`; +} + +function internalUrl(path: string): string { + return `https://void.live${path}`; +} + +class LiveResponseError extends Error { + constructor( + readonly code: LiveControlErrorCode, + message: string, + readonly status: number, + ) { + super(message); + } +} + +function liveResponseError(code: LiveControlErrorCode, message: string, status: number): never { + throw new LiveResponseError(code, message, status); +} + +function errorToResponse(error: unknown): Response { + if (error instanceof LiveResponseError) { + return Response.json( + { + accepted: false, + error: { code: error.code, message: error.message }, + } satisfies LiveControlResponse, + { status: error.status }, + ); + } + return Response.json( + { + accepted: false, + error: { + code: 'INTERNAL_ERROR', + message: error instanceof Error ? error.message : String(error), + }, + } satisfies LiveControlResponse, + { status: 500 }, + ); +} + +async function responseToControlError(response: Response): Promise { + const body = await response.text().catch(() => ''); + const message = body || response.statusText; + if (response.status === 403 || response.status === 401) { + return { code: 'FORBIDDEN', message: message || 'Forbidden' }; + } + return { code: 'BAD_REQUEST', message: message || 'Subscription rejected' }; +} diff --git a/packages/void/test/unit/compile.test.ts b/packages/void/test/unit/compile.test.ts index 10b99e4e6..e4b48830e 100644 --- a/packages/void/test/unit/compile.test.ts +++ b/packages/void/test/unit/compile.test.ts @@ -525,6 +525,27 @@ describe('compileRouteTable', () => { expect(code).toContain('__forwardWebSocketRequest'); }); + it('generates the shared live durable object export when live is enabled', () => { + const scan: ScanResult = { + routes: [], + middleware: [], + jobs: [], + queues: [], + pages: { pages: [], layouts: [], namedLayouts: [] }, + }; + + const code = compileRouteTable(scan, '/routes', '/middleware', '/crons', '/queues', null, { + ...testOptions, + hasLive: true, + liveRuntimePath: '/resolved/void/runtime/live-server.mjs', + }); + + expect(code).toContain('createLiveDurableObject'); + expect(code).toContain( + 'export const VoidLiveStreamDurableObject = __createLiveDurableObject();', + ); + }); + it('wires c.rewrite, c.originalUrl, and c.isRewritten to a shared WeakMap', () => { // Lightweight anchor that the rewrite middleware is emitted at all — // behavioral coverage for the wrapper lives in `c.rewrite runtime wrapper` @@ -2183,6 +2204,46 @@ describe('compileWorkerEntry', () => { expect(code).toContain('const ws = await __voidHandleWebSocket(request, env);'); }); + it('re-exports the live Durable Object class when enabled', () => { + const code = compileWorkerEntry( + false, + undefined, + 'assets', + [], + 'void/_env', + undefined, + false, + false, + true, + ); + + expect(code).toContain('VoidLiveStreamDurableObject'); + expect(code).toContain('export { VoidLiveStreamDurableObject };'); + expect(code).toContain('__setLiveBindingName("VOID_LIVE");'); + }); + + it('re-exports the configured live Durable Object class', () => { + const code = compileWorkerEntry( + false, + undefined, + 'assets', + [], + 'void/_env', + undefined, + false, + false, + { + bindingName: 'VOID_LIVE_abcd1234', + className: 'VoidLiveStreamDurableObject_abcd1234', + }, + ); + + expect(code).toContain( + 'export { VoidLiveStreamDurableObject as VoidLiveStreamDurableObject_abcd1234 };', + ); + expect(code).toContain('__setLiveBindingName("VOID_LIVE_abcd1234");'); + }); + it('re-exports the Cloudflare Sandbox Durable Object class when enabled', () => { const code = compileWorkerEntry(false, undefined, 'assets', [], 'void/_env', undefined, true); diff --git a/packages/void/test/unit/deploy-project.test.ts b/packages/void/test/unit/deploy-project.test.ts index 32ca18ec8..8c2805eea 100644 --- a/packages/void/test/unit/deploy-project.test.ts +++ b/packages/void/test/unit/deploy-project.test.ts @@ -1,9 +1,11 @@ -import { existsSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'pathe'; import { afterEach, describe, expect, it } from 'vite-plus/test'; -import type { ProjectResponse } from '../../src/cli/client.ts'; +import type { PlatformClient, ProjectResponse } from '../../src/cli/client.ts'; import { + ensurePreviouslyDeployedLiveBindingConfig, + resolveDeployLiveBindingConfig, resolveDistDir, resolveDeployWranglerCompat, resolveStaticBuildCommand, @@ -168,6 +170,151 @@ describe('resolveDeployWranglerCompat', () => { }); }); +describe('ensurePreviouslyDeployedLiveBindingConfig', () => { + afterEach(() => { + while (tempDirs.length > 0) { + const dir = tempDirs.pop()!; + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('restores the active deployment live binding into wrangler config', async () => { + const root = mkdtempSync(join(tmpdir(), 'void-test-')); + tempDirs.push(root); + const client = { + async listDeployments() { + return { + activeDeploymentId: 'dpl_live', + deployments: [ + { + id: 'dpl_live', + status: 'live', + source: null, + commit: null, + manifest: JSON.stringify({ + version: 3, + bindings: {}, + live: { + bindingName: 'VOID_LIVE_a1b2c3d4', + className: 'VoidLiveStreamDurableObject_a1b2c3d4', + storage: 'sqlite', + classes: ['VoidLiveStreamDurableObject_a1b2c3d4'], + }, + }), + error_message: null, + retained: true, + applied_migrations: null, + created_at: '2026-05-12T00:00:00.000Z', + deployed_at: '2026-05-12T00:00:00.000Z', + }, + ], + }; + }, + } as unknown as PlatformClient; + + await expect(ensurePreviouslyDeployedLiveBindingConfig(root, client, 'proj_1')).resolves.toBe( + true, + ); + + const wrangler = JSON.parse(readFileSync(join(root, 'wrangler.jsonc'), 'utf-8')); + expect(wrangler.durable_objects.bindings).toContainEqual({ + name: 'VOID_LIVE_a1b2c3d4', + class_name: 'VoidLiveStreamDurableObject_a1b2c3d4', + }); + expect(wrangler.migrations).toContainEqual({ + tag: 'live-v1', + new_sqlite_classes: ['VoidLiveStreamDurableObject_a1b2c3d4'], + }); + }); +}); + +describe('resolveDeployLiveBindingConfig', () => { + afterEach(() => { + while (tempDirs.length > 0) { + const dir = tempDirs.pop()!; + rmSync(dir, { recursive: true, force: true }); + } + }); + + it('does not restore or activate live bindings when current source no longer needs live', async () => { + const root = mkdtempSync(join(tmpdir(), 'void-test-')); + tempDirs.push(root); + writeFileSync( + join(root, 'wrangler.jsonc'), + JSON.stringify({ + durable_objects: { + bindings: [ + { + name: 'VOID_LIVE_a1b2c3d4', + class_name: 'VoidLiveStreamDurableObject_a1b2c3d4', + }, + ], + }, + }), + ); + const client = { + async listDeployments() { + throw new Error('listDeployments should not be called'); + }, + } as unknown as PlatformClient; + + await expect( + resolveDeployLiveBindingConfig(root, client, 'proj_1', { + isFrameworkMode: false, + needsLive: false, + webSockets: [], + }), + ).resolves.toBeUndefined(); + + const wrangler = JSON.parse(readFileSync(join(root, 'wrangler.jsonc'), 'utf-8')); + expect(wrangler.migrations).toBeUndefined(); + }); + + it('restores the active deployment live binding only when current source still needs live', async () => { + const root = mkdtempSync(join(tmpdir(), 'void-test-')); + tempDirs.push(root); + const client = { + async listDeployments() { + return { + activeDeploymentId: 'dpl_live', + deployments: [ + { + id: 'dpl_live', + status: 'live', + source: null, + commit: null, + manifest: JSON.stringify({ + version: 3, + bindings: {}, + live: { + bindingName: 'VOID_LIVE_a1b2c3d4', + className: 'VoidLiveStreamDurableObject_a1b2c3d4', + }, + }), + error_message: null, + retained: true, + applied_migrations: null, + created_at: '2026-05-12T00:00:00.000Z', + deployed_at: '2026-05-12T00:00:00.000Z', + }, + ], + }; + }, + } as unknown as PlatformClient; + + await expect( + resolveDeployLiveBindingConfig(root, client, 'proj_1', { + isFrameworkMode: false, + needsLive: true, + webSockets: [], + }), + ).resolves.toEqual({ + bindingName: 'VOID_LIVE_a1b2c3d4', + className: 'VoidLiveStreamDurableObject_a1b2c3d4', + }); + }); +}); + describe('resolveStaticBuildCommand', () => { it('uses inference.build when provided', () => { expect( diff --git a/packages/void/test/unit/inference.test.ts b/packages/void/test/unit/inference.test.ts index a98279b89..80944b09b 100644 --- a/packages/void/test/unit/inference.test.ts +++ b/packages/void/test/unit/inference.test.ts @@ -305,6 +305,53 @@ describe('inferBindingsFromSource', () => { const result = inferBindingsFromSource(code, 'test.ts'); expect(result.needsAI).toBe(false); }); + + it("detects non-type imports from 'void/live'", () => { + const code = ` + import { defineLiveStream } from "void/live"; + export const live = defineLiveStream({ id: "app" }); + `; + const result = inferBindingsFromSource(code, 'live.ts'); + expect(result.needsLive).toBe(true); + }); + + it("does not detect live infrastructure from 'void/live/client'", () => { + const code = ` + import { connectLiveStream } from "void/live/client"; + connectLiveStream("/live"); + `; + const result = inferBindingsFromSource(code, 'client.ts'); + expect(result.needsLive).toBe(false); + }); + + it("does not detect live infrastructure from type-only 'void/live' imports", () => { + const code = ` + import type { LiveClientEvent } from "void/live"; + export type Event = LiveClientEvent<{ value: string }>; + `; + const result = inferBindingsFromSource(code, 'types.ts'); + expect(result.needsLive).toBe(false); + }); + + it("does not detect live infrastructure from inline type-only 'void/live' imports", () => { + const code = ` + import { type LiveStream, type LiveClientEvent } from "void/live"; + export type Stream = LiveStream; + export type Event = LiveClientEvent<{ value: string }>; + `; + const result = inferBindingsFromSource(code, 'types.ts'); + expect(result.needsLive).toBe(false); + }); + + it("detects live infrastructure from mixed inline type and value 'void/live' imports", () => { + const code = ` + import { type LiveStream, defineLiveStream } from "void/live"; + export type Stream = LiveStream; + export const live = defineLiveStream({ id: "app" }); + `; + const result = inferBindingsFromSource(code, 'live.ts'); + expect(result.needsLive).toBe(true); + }); }); describe('mergeBindings', () => { @@ -360,6 +407,11 @@ describe('mergeBindings', () => { expect(result.needsSandbox).toBe(true); }); + it('merges live flag', () => { + const result = mergeBindings([empty, { ...empty, needsLive: true }]); + expect(result.needsLive).toBe(true); + }); + it('merges AI with other bindings', () => { const result = mergeBindings([ { ...empty, needsD1: true }, diff --git a/packages/void/test/unit/live-client.test.ts b/packages/void/test/unit/live-client.test.ts new file mode 100644 index 000000000..c58499898 --- /dev/null +++ b/packages/void/test/unit/live-client.test.ts @@ -0,0 +1,614 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vite-plus/test'; +import { connectLiveStream } from '../../src/runtime/live-client.ts'; + +class MockEventSource { + static instances: Array = []; + + listeners = new Map void>>(); + closed = false; + + constructor( + readonly url: string, + readonly options?: { withCredentials?: boolean }, + ) { + MockEventSource.instances.push(this); + } + + addEventListener(type: string, listener: (event: Event) => void): void { + const listeners = this.listeners.get(type) ?? new Set(); + listeners.add(listener); + this.listeners.set(type, listeners); + } + + removeEventListener(type: string, listener: (event: Event) => void): void { + this.listeners.get(type)?.delete(listener); + } + + emit(type: string, event: Event): void { + for (const listener of this.listeners.get(type) ?? []) { + listener(event); + } + } + + close(): void { + this.closed = true; + } +} + +const originalLocation = (globalThis as { location?: unknown }).location; + +describe('connectLiveStream', () => { + beforeEach(() => { + MockEventSource.instances = []; + vi.stubGlobal('location', { href: 'https://void.test/page' }); + }); + + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); + (globalThis as { location?: unknown }).location = originalLocation; + }); + + it('opens one EventSource and sends subscribe after open', async () => { + const fetchMock = vi.fn(async () => + Response.json({ + accepted: true, + connectionId: 'connection', + results: [{ ok: true, kind: 'subscribe', id: 'post-card', topic: 'post:12' }], + }), + ); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + withCredentials: true, + }); + + const pending = live.subscribe({ id: 'post-card', topic: 'post:12' }); + await Promise.resolve(); + expect(fetchMock).not.toHaveBeenCalled(); + const source = MockEventSource.instances[0]!; + expect(source.url).toContain('/live?connectionId='); + expect(source.options?.withCredentials).toBe(true); + + source.emit('open', new Event('open')); + await pending; + + expect(fetchMock).toHaveBeenCalledWith('/live', { + method: 'POST', + headers: expect.any(Headers), + credentials: 'include', + body: expect.stringContaining('"topic":"post:12"'), + }); + }); + + it('rejects a subscribe waiting for the initial open when closed', async () => { + const fetchMock = vi.fn(async () => + Response.json({ + accepted: true, + connectionId: 'connection', + results: [{ ok: true, kind: 'subscribe', id: 'post-card', topic: 'post:12' }], + }), + ); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + }); + + const pending = live.subscribe({ id: 'post-card', topic: 'post:12' }); + await Promise.resolve(); + const source = MockEventSource.instances[0]!; + + live.close(); + + await expect(pending).rejects.toThrow('live: stream is closed.'); + expect(source.closed).toBe(true); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it('routes message envelopes by subscription id and unsubscribes individually', async () => { + const fetchMock = vi.fn(async (_url, init) => { + const body = JSON.parse(String((init as RequestInit).body)); + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: body.operations.map((op: { kind: string; id: string; topic?: string }) => ({ + ok: true, + kind: op.kind, + id: op.id, + topic: op.topic, + })), + }); + }); + const handler = vi.fn(); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + }); + const unsubscribe = live.subscribe({ id: 'comments', topic: 'comment:2323', onEvent: handler }); + await Promise.resolve(); + const source = MockEventSource.instances[0]!; + source.emit('open', new Event('open')); + await unsubscribe; + + source.emit('message', { + data: JSON.stringify({ + subscriptionId: 'comments', + topic: 'comment:2323', + data: { ok: true }, + }), + } as MessageEvent); + expect(handler).toHaveBeenCalledWith({ + subscriptionId: 'comments', + topic: 'comment:2323', + data: { ok: true }, + }); + + await ( + await unsubscribe + )(); + expect(fetchMock).toHaveBeenLastCalledWith('/live', { + method: 'POST', + headers: expect.any(Headers), + credentials: 'same-origin', + body: expect.stringContaining('"kind":"unsubscribe"'), + }); + }); + + it('routes events that arrive before the subscribe control ack resolves', async () => { + const handler = vi.fn(); + const fetchMock = vi.fn(async (_url, init) => { + const body = JSON.parse(String((init as RequestInit).body)); + MockEventSource.instances[0]!.emit('message', { + data: JSON.stringify({ + subscriptionId: 'comments', + topic: 'comment:2323', + data: { early: true }, + }), + } as MessageEvent); + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: [{ ok: true, kind: 'subscribe', id: 'comments', topic: 'comment:2323' }], + }); + }); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + }); + + const unsubscribe = live.subscribe({ id: 'comments', topic: 'comment:2323', onEvent: handler }); + await Promise.resolve(); + MockEventSource.instances[0]!.emit('open', new Event('open')); + await unsubscribe; + + expect(handler).toHaveBeenCalledWith({ + subscriptionId: 'comments', + topic: 'comment:2323', + data: { early: true }, + }); + }); + + it('routes replacement events to the replacement handler before the control ack resolves', async () => { + const oldHandler = vi.fn(); + const newHandler = vi.fn(); + const fetchMock = vi.fn(async (_url, init) => { + const body = JSON.parse(String((init as RequestInit).body)); + if (body.operations[0]?.topic === 'post:13') { + MockEventSource.instances[0]!.emit('message', { + data: JSON.stringify({ + subscriptionId: 'card', + topic: 'post:13', + data: { early: true }, + }), + } as MessageEvent); + } + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: body.operations.map((op: { kind: string; id: string; topic?: string }) => ({ + ok: true, + kind: op.kind, + id: op.id, + topic: op.topic, + })), + }); + }); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + }); + + const first = live.subscribe({ id: 'card', topic: 'post:12', onEvent: oldHandler }); + await Promise.resolve(); + MockEventSource.instances[0]!.emit('open', new Event('open')); + await first; + + const second = live.subscribe({ id: 'card', topic: 'post:13', onEvent: newHandler }); + await second; + + expect(oldHandler).not.toHaveBeenCalled(); + expect(newHandler).toHaveBeenCalledWith({ + subscriptionId: 'card', + topic: 'post:13', + data: { early: true }, + }); + }); + + it('keeps per-operation subscribe failures isolated in one control batch', async () => { + const fetchMock = vi.fn(async (_url, init) => { + const body = JSON.parse(String((init as RequestInit).body)); + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: [ + { ok: true, kind: 'subscribe', id: 'post-card', topic: 'post:12' }, + { + ok: false, + kind: 'subscribe', + id: 'private-card', + error: { code: 'FORBIDDEN', message: 'Forbidden' }, + }, + ], + }); + }); + const postHandler = vi.fn(); + const privateHandler = vi.fn(); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + }); + + const post = live.subscribe({ id: 'post-card', topic: 'post:12', onEvent: postHandler }); + const privatePost = live.subscribe({ + id: 'private-card', + topic: 'post:private', + onEvent: privateHandler, + }); + await Promise.resolve(); + MockEventSource.instances[0]!.emit('open', new Event('open')); + + await expect(post).resolves.toEqual(expect.any(Function)); + await expect(privatePost).rejects.toThrow('Forbidden'); + + MockEventSource.instances[0]!.emit('message', { + data: JSON.stringify({ subscriptionId: 'post-card', topic: 'post:12', data: { ok: true } }), + } as MessageEvent); + MockEventSource.instances[0]!.emit('message', { + data: JSON.stringify({ subscriptionId: 'private-card', topic: 'post:private', data: {} }), + } as MessageEvent); + + expect(postHandler).toHaveBeenCalledTimes(1); + expect(privateHandler).not.toHaveBeenCalled(); + }); + + it('keeps the previous subscription handler when replacement fails', async () => { + const fetchMock = vi.fn(async (_url, init) => { + const body = JSON.parse(String((init as RequestInit).body)); + const operation = body.operations[0]; + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: + operation.topic === 'post:blocked' + ? [ + { + ok: false, + kind: 'subscribe', + id: operation.id, + error: { code: 'FORBIDDEN', message: 'Forbidden' }, + }, + ] + : [{ ok: true, kind: operation.kind, id: operation.id, topic: operation.topic }], + }); + }); + const oldHandler = vi.fn(); + const newHandler = vi.fn(); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + }); + + const first = live.subscribe({ id: 'card', topic: 'post:12', onEvent: oldHandler }); + await Promise.resolve(); + MockEventSource.instances[0]!.emit('open', new Event('open')); + await first; + + await expect( + live.subscribe({ id: 'card', topic: 'post:blocked', onEvent: newHandler }), + ).rejects.toThrow('Forbidden'); + + MockEventSource.instances[0]!.emit('message', { + data: JSON.stringify({ subscriptionId: 'card', topic: 'post:12', data: { ok: true } }), + } as MessageEvent); + + expect(oldHandler).toHaveBeenCalledTimes(1); + expect(newHandler).not.toHaveBeenCalled(); + }); + + it('does not let stale unsubscribe functions remove replacement subscriptions', async () => { + const fetchMock = vi.fn(async (_url, init) => { + const body = JSON.parse(String((init as RequestInit).body)); + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: body.operations.map((op: { kind: string; id: string; topic?: string }) => ({ + ok: true, + kind: op.kind, + id: op.id, + topic: op.topic, + })), + }); + }); + const oldHandler = vi.fn(); + const newHandler = vi.fn(); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + }); + + const firstUnsubscribe = live.subscribe({ + id: 'card', + topic: 'post:12', + onEvent: oldHandler, + }); + await Promise.resolve(); + MockEventSource.instances[0]!.emit('open', new Event('open')); + const unsubscribeOld = await firstUnsubscribe; + await flushMicrotasks(); + + const secondUnsubscribe = live.subscribe({ + id: 'card', + topic: 'post:13', + onEvent: newHandler, + }); + const unsubscribeNew = await secondUnsubscribe; + expect(fetchMock).toHaveBeenCalledTimes(2); + + await unsubscribeOld(); + expect(fetchMock).toHaveBeenCalledTimes(2); + + MockEventSource.instances[0]!.emit('message', { + data: JSON.stringify({ subscriptionId: 'card', topic: 'post:13', data: { ok: true } }), + } as MessageEvent); + expect(oldHandler).not.toHaveBeenCalled(); + expect(newHandler).toHaveBeenCalledTimes(1); + + await unsubscribeNew(); + expect(fetchMock).toHaveBeenCalledTimes(3); + }); + + it('does not let stale unsubscribe functions race in-flight replacements', async () => { + let resolveReplacement: ((value: Response | PromiseLike) => void) | undefined; + const fetchMock = vi.fn(async (_url, init) => { + const body = JSON.parse(String((init as RequestInit).body)); + if (body.operations[0]?.topic === 'post:13') { + return await new Promise((resolve) => { + resolveReplacement = resolve; + }); + } + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: body.operations.map((op: { kind: string; id: string; topic?: string }) => ({ + ok: true, + kind: op.kind, + id: op.id, + topic: op.topic, + })), + }); + }); + const oldHandler = vi.fn(); + const newHandler = vi.fn(); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + }); + + const firstUnsubscribe = live.subscribe({ + id: 'card', + topic: 'post:12', + onEvent: oldHandler, + }); + await Promise.resolve(); + MockEventSource.instances[0]!.emit('open', new Event('open')); + const unsubscribeOld = await firstUnsubscribe; + + const secondUnsubscribe = live.subscribe({ + id: 'card', + topic: 'post:13', + onEvent: newHandler, + }); + await flushMicrotasks(); + expect(fetchMock).toHaveBeenCalledTimes(2); + + await unsubscribeOld(); + expect(fetchMock).toHaveBeenCalledTimes(2); + + resolveReplacement!( + Response.json({ + accepted: true, + connectionId: 'connection', + results: [{ ok: true, kind: 'subscribe', id: 'card', topic: 'post:13' }], + }), + ); + const unsubscribeNew = await secondUnsubscribe; + + MockEventSource.instances[0]!.emit('message', { + data: JSON.stringify({ subscriptionId: 'card', topic: 'post:13', data: { ok: true } }), + } as MessageEvent); + expect(oldHandler).not.toHaveBeenCalled(); + expect(newHandler).toHaveBeenCalledTimes(1); + + await unsubscribeNew(); + expect(fetchMock).toHaveBeenCalledTimes(3); + }); + + it('serializes control requests while a previous flush is in flight', async () => { + let resolveReplacement: ((value: Response | PromiseLike) => void) | undefined; + const fetchMock = vi.fn(async (_url, init) => { + const body = JSON.parse(String((init as RequestInit).body)); + if (body.operations[0]?.topic === 'post:13') { + return await new Promise((resolve) => { + resolveReplacement = resolve; + }); + } + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: body.operations.map((op: { kind: string; id: string; topic?: string }) => ({ + ok: true, + kind: op.kind, + id: op.id, + topic: op.topic, + })), + }); + }); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + }); + + const first = live.subscribe({ id: 'card', topic: 'post:12' }); + await Promise.resolve(); + MockEventSource.instances[0]!.emit('open', new Event('open')); + await first; + await flushMicrotasks(); + + const replacement = live.subscribe({ id: 'card', topic: 'post:13' }); + await Promise.resolve(); + await Promise.resolve(); + expect(fetchMock).toHaveBeenCalledTimes(2); + + const newest = live.subscribe({ id: 'card', topic: 'post:14' }); + await Promise.resolve(); + await Promise.resolve(); + expect(fetchMock).toHaveBeenCalledTimes(2); + + resolveReplacement!( + Response.json({ + accepted: true, + connectionId: 'connection', + results: [{ ok: true, kind: 'subscribe', id: 'card', topic: 'post:13' }], + }), + ); + await replacement; + await newest; + + expect(fetchMock).toHaveBeenCalledTimes(3); + expect(String(fetchMock.mock.calls[1]![1]?.body)).toContain('"topic":"post:13"'); + expect(String(fetchMock.mock.calls[2]![1]?.body)).toContain('"topic":"post:14"'); + }); + + it('reconnects and resubscribes active subscriptions with lastEventId', async () => { + vi.useFakeTimers(); + const fetchMock = vi.fn(async (_url, init) => { + const body = JSON.parse(String((init as RequestInit).body)); + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: body.operations.map((op: { kind: string; id: string; topic?: string }) => ({ + ok: true, + kind: op.kind, + id: op.id, + topic: op.topic, + })), + }); + }); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + retryDelay: 25, + }); + const unsubscribe = live.subscribe({ id: 'post-card', topic: 'post:12', lastEventId: 'v1' }); + await Promise.resolve(); + const firstSource = MockEventSource.instances[0]!; + firstSource.emit('open', new Event('open')); + await unsubscribe; + + firstSource.emit('message', { + data: JSON.stringify({ + subscriptionId: 'post-card', + topic: 'post:12', + eventId: 'v2', + data: {}, + }), + } as MessageEvent); + firstSource.emit('error', new Event('error')); + expect(firstSource.closed).toBe(true); + + await vi.advanceTimersByTimeAsync(25); + const secondSource = MockEventSource.instances[1]!; + secondSource.emit('open', new Event('open')); + await Promise.resolve(); + await Promise.resolve(); + + expect(fetchMock).toHaveBeenLastCalledWith('/live', { + method: 'POST', + headers: expect.any(Headers), + credentials: 'same-origin', + body: expect.stringContaining('"lastEventId":"v2"'), + }); + }); + + it('preserves resubscribe when a control operation interrupts a reconnect timer', async () => { + vi.useFakeTimers(); + const fetchMock = vi.fn(async (_url, init) => { + const body = JSON.parse(String((init as RequestInit).body)); + return Response.json({ + accepted: true, + connectionId: body.connectionId, + results: body.operations.map((op: { kind: string; id: string; topic?: string }) => ({ + ok: true, + kind: op.kind, + id: op.id, + topic: op.topic, + })), + }); + }); + const live = connectLiveStream('/live', { + eventSource: MockEventSource, + fetch: fetchMock as unknown as typeof fetch, + retryDelay: 25, + }); + const first = live.subscribe({ id: 'post-card', topic: 'post:12', lastEventId: 'v1' }); + await Promise.resolve(); + const firstSource = MockEventSource.instances[0]!; + firstSource.emit('open', new Event('open')); + await first; + await flushMicrotasks(); + + firstSource.emit('message', { + data: JSON.stringify({ + subscriptionId: 'post-card', + topic: 'post:12', + eventId: 'v2', + data: {}, + }), + } as MessageEvent); + firstSource.emit('error', new Event('error')); + + const second = live.subscribe({ id: 'comments', topic: 'comment:2323' }); + await flushMicrotasks(); + const secondSource = MockEventSource.instances[1]!; + secondSource.emit('open', new Event('open')); + await second; + + expect(fetchMock).toHaveBeenLastCalledWith('/live', { + method: 'POST', + headers: expect.any(Headers), + credentials: 'same-origin', + body: expect.stringContaining('"id":"post-card"'), + }); + expect(String(fetchMock.mock.calls.at(-1)?.[1]?.body)).toContain('"id":"comments"'); + expect(String(fetchMock.mock.calls.at(-1)?.[1]?.body)).toContain('"lastEventId":"v2"'); + + await vi.advanceTimersByTimeAsync(25); + expect(MockEventSource.instances).toHaveLength(2); + }); +}); + +async function flushMicrotasks(count = 5): Promise { + for (let i = 0; i < count; i++) { + await Promise.resolve(); + } +} diff --git a/packages/void/test/unit/live-server.test.ts b/packages/void/test/unit/live-server.test.ts new file mode 100644 index 000000000..f30aced47 --- /dev/null +++ b/packages/void/test/unit/live-server.test.ts @@ -0,0 +1,600 @@ +import { describe, expect, it } from 'vite-plus/test'; +import type { CloudContext } from '../../src/runtime/handler.ts'; +import { createLiveDurableObject } from '../../src/runtime/live-server.ts'; +import { defineLiveStream, type LiveLimits } from '../../src/runtime/live.ts'; + +const LIMITS: LiveLimits = { + maxSubscriptionsPerConnection: 256, + maxSubscriptionsPerTopic: 64, + maxOperationsPerControlRequest: 100, + maxQueuedEventsPerConnection: 100, + maxEncodedEventSize: 64 * 1024, + deliveryAttemptTimeoutMs: 1_500, +}; + +describe('live Durable Object runtime', () => { + it('rejects duplicate stream ids at startup', () => { + const id = `duplicate-${crypto.randomUUID()}`; + defineLiveStream({ id, allowAnonymousControl: true }); + expect(() => defineLiveStream({ id, allowAnonymousControl: true })).toThrow( + `live: duplicate stream id "${id}".`, + ); + }); + + it('rejects invalid live limits before registering the stream id', () => { + const id = `limits-${crypto.randomUUID()}`; + + expect(() => + defineLiveStream({ + id, + allowAnonymousControl: true, + limits: { maxQueuedEventsPerConnection: Number.NaN }, + }), + ).toThrow('limits.maxQueuedEventsPerConnection must be a positive finite integer'); + expect(() => + defineLiveStream({ + id, + allowAnonymousControl: true, + limits: { maxSubscriptionsPerTopic: 257 }, + }), + ).toThrow('limits.maxSubscriptionsPerTopic must be at most 256'); + expect(() => defineLiveStream({ id, allowAnonymousControl: true })).not.toThrow(); + }); + + it('returns per-operation errors for invalid control operations', async () => { + const harness = createHarness(); + const stream = defineLiveStream({ + id: `control-${crypto.randomUUID()}`, + allowAnonymousControl: true, + }); + const connectionId = 'connection-control-1'; + const connectResponse = await stream.connect( + liveContext(harness.env, `https://void.live/live?connectionId=${connectionId}`), + ); + const reader = connectResponse.body!.getReader(); + await readFrame(reader); + + const controlResponse = await stream.control( + liveContext(harness.env, 'https://void.live/live', { + connectionId, + operations: [ + { kind: 'subscribe', id: 'valid-card', topic: 'post:12' }, + { kind: 'subscribe', id: 'bad-topic', topic: '' }, + { kind: 'unsubscribe', id: 'bad\nid' }, + ], + }), + ); + const body = await controlResponse.json(); + + expect(controlResponse.ok).toBe(true); + expect(body).toMatchObject({ + accepted: true, + results: [ + { ok: true, kind: 'subscribe', id: 'valid-card', topic: 'post:12' }, + { ok: false, kind: 'subscribe', id: 'bad-topic', error: { code: 'INVALID_TOPIC' } }, + { ok: false, kind: 'unsubscribe', error: { code: 'INVALID_SUBSCRIPTION_ID' } }, + ], + }); + await reader.cancel(); + }); + + it('prunes stale topic rows before enforcing the topic cap', async () => { + const harness = createHarness(); + const topic = harness.stub('live:app:topic:topic-key'); + const limits = { ...LIMITS, maxSubscriptionsPerTopic: 1 }; + + const stale = await topic.fetch( + jsonRequest('/register', { + streamId: 'app', + topic: 'post:12', + topicKey: 'topic-key', + connectionId: 'stale-connection-0001', + subscriptionId: 'comments', + generation: 1, + revision: 1, + lastSeenAt: Date.now(), + limits, + }), + ); + expect(stale.ok).toBe(true); + + const current = await topic.fetch( + jsonRequest('/register', { + streamId: 'app', + topic: 'post:12', + topicKey: 'topic-key', + connectionId: 'current-connection-001', + subscriptionId: 'comments', + generation: 1, + revision: 1, + lastSeenAt: Date.now(), + limits, + }), + ); + expect(current.ok).toBe(true); + }); + + it('bounds publish delivery attempts', async () => { + const LiveDurableObject = createLiveDurableObject(); + const state = createState(); + const env = { + VOID_LIVE: { + idFromName(name: string) { + return name as unknown as DurableObjectId; + }, + get() { + return { + fetch() { + return new Promise(() => {}); + }, + } as unknown as DurableObjectStub; + }, + } as unknown as DurableObjectNamespace, + }; + const topic = new LiveDurableObject(state, env); + const limits = { ...LIMITS, deliveryAttemptTimeoutMs: 10 }; + + await topic.fetch( + jsonRequest('/register', { + streamId: 'app', + topic: 'post:12', + topicKey: 'topic-key', + connectionId: 'slow-connection-0001', + subscriptionId: 'comments', + generation: 1, + revision: 1, + lastSeenAt: Date.now(), + limits, + }), + ); + + const startedAt = Date.now(); + const response = await topic.fetch( + jsonRequest('/publish', { + streamId: 'app', + topic: 'post:12', + topicKey: 'topic-key', + data: { body: 'hello' }, + limits, + }), + ); + + expect(response.ok).toBe(true); + expect(Date.now() - startedAt).toBeLessThan(1_000); + }); + + it('serializes publishes for one topic so slow stale connections cannot reorder events', async () => { + let slowCalls = 0; + let resolveSlowFirst: ((response: Response) => void) | undefined; + let markSlowFirstStarted: (() => void) | undefined; + const slowFirstStarted = new Promise((resolve) => { + markSlowFirstStarted = resolve; + }); + const harness = createHarness({ + fetch(name, request) { + if ( + name === 'live:app:connection:slow-connection-0001' && + new URL(request.url).pathname === '/deliver' + ) { + slowCalls++; + if (slowCalls === 1) { + return new Promise((resolve) => { + markSlowFirstStarted!(); + resolveSlowFirst = resolve; + }); + } + return Promise.resolve(new Response('connection missing', { status: 410 })); + } + }, + }); + const topicKey = await topicDigest('app', 'post:12'); + const topic = harness.stub(`live:app:topic:${topicKey}`); + await topic.fetch( + jsonRequest('/register', { + streamId: 'app', + topic: 'post:12', + topicKey, + connectionId: 'slow-connection-0001', + subscriptionId: 'slow-card', + generation: 1, + revision: 1, + lastSeenAt: Date.now(), + limits: LIMITS, + }), + ); + + const connectionId = 'connection-order-001'; + const connection = harness.stub(`live:app:connection:${connectionId}`); + const response = await connection.fetch( + jsonRequest('/connect', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits: LIMITS, + }), + ); + const reader = response.body!.getReader(); + await readFrame(reader); + await connection.fetch( + jsonRequest('/control', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits: LIMITS, + operations: [{ kind: 'subscribe', id: 'post-card', topic: 'post:12' }], + }), + ); + + const firstPublish = harness.publish('app', 'post:12', { seq: 1 }); + await slowFirstStarted; + const secondPublish = harness.publish('app', 'post:12', { seq: 2 }); + await Promise.resolve(); + await Promise.resolve(); + resolveSlowFirst!(new Response('connection missing', { status: 410 })); + + expect(parseData(await readFrame(reader))).toMatchObject({ data: { seq: 1 } }); + expect(parseData(await readFrame(reader))).toMatchObject({ data: { seq: 2 } }); + await Promise.all([firstPublish, secondPublish]); + await reader.cancel(); + }); + + it('rejects oversized live events instead of reporting successful publish', async () => { + const harness = createHarness(); + const connectionId = 'connection-large-001'; + const connection = harness.stub(`live:app:connection:${connectionId}`); + const response = await connection.fetch( + jsonRequest('/connect', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits: LIMITS, + }), + ); + const reader = response.body!.getReader(); + await readFrame(reader); + await connection.fetch( + jsonRequest('/control', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits: LIMITS, + operations: [{ kind: 'subscribe', id: 'post-card', topic: 'post:12' }], + }), + ); + + const publishResponse = await harness.publish( + 'app', + 'post:12', + { body: 'x'.repeat(200) }, + { ...LIMITS, maxEncodedEventSize: 80 }, + ); + const body = await publishResponse.json(); + + expect(publishResponse.status).toBe(413); + expect(body).toMatchObject({ + accepted: false, + error: { code: 'PAYLOAD_TOO_LARGE' }, + }); + await reader.cancel(); + }); + + it('keeps replacement successful when old topic unregister fails', async () => { + const oldTopicKey = await topicDigest('app', 'post:12'); + const harness = createHarness({ + fetch(name, request) { + if ( + name === `live:app:topic:${oldTopicKey}` && + new URL(request.url).pathname === '/unregister' + ) { + return Promise.reject(new Error('temporary unregister failure')); + } + }, + }); + const connectionId = 'connection-replace-1'; + const connection = harness.stub(`live:app:connection:${connectionId}`); + const response = await connection.fetch( + jsonRequest('/connect', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits: LIMITS, + }), + ); + const reader = response.body!.getReader(); + await readFrame(reader); + await connection.fetch( + jsonRequest('/control', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits: LIMITS, + operations: [{ kind: 'subscribe', id: 'post-card', topic: 'post:12' }], + }), + ); + + const replaceResponse = await connection.fetch( + jsonRequest('/control', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits: LIMITS, + operations: [{ kind: 'subscribe', id: 'post-card', topic: 'post:13' }], + }), + ); + await expect(replaceResponse.json()).resolves.toMatchObject({ + accepted: true, + results: [{ ok: true, kind: 'subscribe', id: 'post-card', topic: 'post:13' }], + }); + + await harness.publish('app', 'post:13', { title: 'Replacement' }); + expect(parseData(await readFrame(reader))).toMatchObject({ + subscriptionId: 'post-card', + topic: 'post:13', + data: { title: 'Replacement' }, + }); + await reader.cancel(); + }); + + it('delivers multiple topics over one connection and unsubscribes individually', async () => { + const harness = createHarness(); + const connectionId = 'connection-0000000001'; + const connection = harness.stub(`live:app:connection:${connectionId}`); + const response = await connection.fetch( + jsonRequest('/connect', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits: LIMITS, + }), + ); + const reader = response.body!.getReader(); + await readFrame(reader); + + await connection.fetch( + jsonRequest('/control', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits: LIMITS, + operations: [ + { kind: 'subscribe', id: 'post-card', topic: 'post:12' }, + { kind: 'subscribe', id: 'comments', topic: 'comment:2323' }, + ], + }), + ); + + const postPublish = harness.publish('app', 'post:12', { title: 'Updated' }); + + expect(parseData(await readFrame(reader))).toMatchObject({ + subscriptionId: 'post-card', + topic: 'post:12', + data: { title: 'Updated' }, + }); + await postPublish; + + const commentPublish = harness.publish('app', 'comment:2323', { count: 4 }); + expect(parseData(await readFrame(reader))).toMatchObject({ + subscriptionId: 'comments', + topic: 'comment:2323', + data: { count: 4 }, + }); + await commentPublish; + + await connection.fetch( + jsonRequest('/control', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits: LIMITS, + operations: [{ kind: 'unsubscribe', id: 'post-card' }], + }), + ); + + await harness.publish('app', 'post:12', { title: 'Ignored' }); + const nextCommentPublish = harness.publish('app', 'comment:2323', { count: 5 }); + + expect(parseData(await readFrame(reader))).toMatchObject({ + subscriptionId: 'comments', + topic: 'comment:2323', + data: { count: 5 }, + }); + await nextCommentPublish; + + await reader.cancel(); + }); + + it('removes previous generation topic rows when a connection reconnects', async () => { + const harness = createHarness(); + const limits = { ...LIMITS, maxSubscriptionsPerTopic: 1 }; + const connectionId = 'connection-0000000002'; + const connection = harness.stub(`live:app:connection:${connectionId}`); + const first = await connection.fetch( + jsonRequest('/connect', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits, + }), + ); + const firstReader = first.body!.getReader(); + await readFrame(firstReader); + + await connection.fetch( + jsonRequest('/control', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits, + operations: [{ kind: 'subscribe', id: 'old-card', topic: 'post:12' }], + }), + ); + + const second = await connection.fetch( + jsonRequest('/connect', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits, + }), + ); + const secondReader = second.body!.getReader(); + await readFrame(secondReader); + + const controlResponse = await connection.fetch( + jsonRequest('/control', { + streamId: 'app', + connectionId, + owner: 'user:1', + limits, + operations: [{ kind: 'subscribe', id: 'new-card', topic: 'post:12' }], + }), + ); + + await expect(controlResponse.json()).resolves.toMatchObject({ + accepted: true, + results: [{ ok: true, id: 'new-card', topic: 'post:12' }], + }); + + await firstReader.cancel().catch(() => {}); + await secondReader.cancel(); + }); +}); + +function createHarness(options?: { + fetch?: (name: string, request: Request) => Promise | undefined; +}) { + const LiveDurableObject = createLiveDurableObject(); + const instances = new Map }>(); + const states = new Map(); + const env = {} as { VOID_LIVE: DurableObjectNamespace }; + const namespace = { + idFromName(name: string) { + return name as unknown as DurableObjectId; + }, + get(id: DurableObjectId) { + const name = id as unknown as string; + return { + fetch(input: RequestInfo | URL, init?: RequestInit) { + const request = input instanceof Request ? input : new Request(input, init); + const response = options?.fetch?.(name, request); + if (response) { + return response; + } + let instance = instances.get(name); + if (!instance) { + const state = createState(); + states.set(name, state); + instance = new LiveDurableObject(state, env); + instances.set(name, instance); + } + return instance.fetch(request); + }, + } as DurableObjectStub; + }, + } as DurableObjectNamespace; + env.VOID_LIVE = namespace; + + return { + env, + stub(name: string): DurableObjectStub { + return namespace.get(namespace.idFromName(name)); + }, + async publish(streamId: string, topic: string, data: unknown, limits: LiveLimits = LIMITS) { + const topicKey = await topicDigest(streamId, topic); + const topicStub = namespace.get(namespace.idFromName(`live:${streamId}:topic:${topicKey}`)); + return topicStub.fetch( + jsonRequest('/publish', { + streamId, + topic, + topicKey, + data, + limits, + }), + ); + }, + }; +} + +function liveContext(env: unknown, url: string, body?: unknown): CloudContext { + return { + env, + req: { + raw: new Request(url, { + method: body === undefined ? 'GET' : 'POST', + body: body === undefined ? undefined : JSON.stringify(body), + }), + }, + get() { + return undefined; + }, + } as unknown as CloudContext; +} + +function createState(): DurableObjectState { + const values = new Map(); + return { + storage: { + get(key: string) { + return Promise.resolve(values.get(key)); + }, + put(key: string, value: unknown) { + values.set(key, value); + return Promise.resolve(); + }, + delete(keyOrKeys: string | Array) { + for (const key of Array.isArray(keyOrKeys) ? keyOrKeys : [keyOrKeys]) { + values.delete(key); + } + return Promise.resolve(); + }, + list(options?: { prefix?: string }) { + const result = new Map(); + for (const [key, value] of values) { + if (!options?.prefix || key.startsWith(options.prefix)) { + result.set(key, value); + } + } + return Promise.resolve(result); + }, + setAlarm() { + return Promise.resolve(); + }, + }, + } as unknown as DurableObjectState; +} + +function jsonRequest(path: string, body: unknown): Request { + return new Request(`https://void.live${path}`, { + method: 'POST', + body: JSON.stringify(body), + }); +} + +async function readFrame(reader: ReadableStreamDefaultReader): Promise { + const decoder = new TextDecoder(); + let frame = ''; + while (!frame.includes('\n\n')) { + const result = await reader.read(); + if (result.done) { + throw new Error('stream ended before frame'); + } + frame += decoder.decode(result.value, { stream: true }); + } + return frame.slice(0, frame.indexOf('\n\n')); +} + +function parseData(frame: string): unknown { + const data = frame + .split('\n') + .filter((line) => line.startsWith('data: ')) + .map((line) => line.slice('data: '.length)) + .join('\n'); + return JSON.parse(data); +} + +async function topicDigest(streamId: string, topic: string): Promise { + const bytes = new TextEncoder().encode(`${streamId}\u0000${topic}`); + const digest = await crypto.subtle.digest('SHA-256', bytes); + const binary = String.fromCharCode(...new Uint8Array(digest)); + return btoa(binary).replaceAll('+', '-').replaceAll('/', '_').replace(/=+$/, ''); +} diff --git a/packages/void/test/unit/migration-binding-consistency.test.ts b/packages/void/test/unit/migration-binding-consistency.test.ts index 9b385c570..213f4d43d 100644 --- a/packages/void/test/unit/migration-binding-consistency.test.ts +++ b/packages/void/test/unit/migration-binding-consistency.test.ts @@ -73,6 +73,40 @@ describe('buildVoidManifest migration-binding consistency', () => { expect(manifest.migrations).toBeUndefined(); }); + it('includes live durable object manifest when void/live is used', () => { + const manifest = buildVoidManifest({ + bindings: { ...noBindings, needsLive: true }, + }); + + expect(manifest.live).toEqual({ + bindingName: 'VOID_LIVE', + className: 'VoidLiveStreamDurableObject', + storage: 'sqlite', + classes: ['VoidLiveStreamDurableObject'], + migrations: [{ tag: 'live-v1', newSqliteClasses: ['VoidLiveStreamDurableObject'] }], + }); + }); + + it('uses resolved live durable object names in the manifest', () => { + const manifest = buildVoidManifest({ + bindings: { ...noBindings, needsLive: true }, + options: { + live: { + bindingName: 'VOID_LIVE_abcd1234', + className: 'VoidLiveStreamDurableObject_abcd1234', + }, + }, + }); + + expect(manifest.live).toEqual({ + bindingName: 'VOID_LIVE_abcd1234', + className: 'VoidLiveStreamDurableObject_abcd1234', + storage: 'sqlite', + classes: ['VoidLiveStreamDurableObject_abcd1234'], + migrations: [{ tag: 'live-v1', newSqliteClasses: ['VoidLiveStreamDurableObject_abcd1234'] }], + }); + }); + it('has no migrations when migrations array is empty', () => { const manifest = buildVoidManifest({ bindings: noBindings, migrations: [] }); @@ -218,6 +252,16 @@ describe('buildFrameworkManifest migration-binding consistency', () => { expect(manifest.bindings.d1).toBeUndefined(); expect(manifest.migrations).toEqual({ dialect: 'postgresql', pending: 1 }); }); + + it('rejects live bindings because framework workers do not export the live durable object', () => { + expect(() => + buildFrameworkManifest({ + frameworkName: 'nuxt', + workerMain: 'index.js', + bindings: { ...noBindings, needsLive: true }, + }), + ).toThrow('void/live is currently supported in Void apps only'); + }); }); // --------------------------------------------------------------------------- diff --git a/packages/void/test/unit/plugin-inference.test.ts b/packages/void/test/unit/plugin-inference.test.ts index 3e891dc8e..dacff13dd 100644 --- a/packages/void/test/unit/plugin-inference.test.ts +++ b/packages/void/test/unit/plugin-inference.test.ts @@ -1,4 +1,4 @@ -import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from 'node:fs'; +import { mkdtempSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'pathe'; import { afterEach, describe, expect, it, vi } from 'vite-plus/test'; @@ -59,6 +59,7 @@ describe('inferProjectBindings', () => { needsQueues: false, needsAI: false, needsSandbox: false, + needsLive: false, }); }); @@ -78,6 +79,30 @@ describe('inferProjectBindings', () => { expect(result.needsAuth).toBe(true); }); + it('detects live from the documented src/live.ts module', () => { + const root = createTempRoot(); + mkdirSync(join(root, 'routes'), { recursive: true }); + writeFileSync( + join(root, 'src', 'live.ts'), + ` + import { defineLiveStream } from "void/live"; + export const live = defineLiveStream({ id: "app" }); + `, + ); + writeFileSync( + join(root, 'routes', 'live.ts'), + ` + import { defineHandler } from "void"; + import { live } from "../src/live"; + export const GET = defineHandler((c) => live.connect(c)); + export const POST = defineHandler((c) => live.control(c)); + `, + ); + + const result = inferProjectBindings(root); + expect(result.needsLive).toBe(true); + }); + it('detects auth from void/auth import in routes/', () => { const root = createTempRoot(); mkdirSync(join(root, 'routes', 'api'), { recursive: true }); @@ -283,6 +308,36 @@ describe('detectFramework', () => { } }); + it('fails client builds that import void/live', () => { + const root = createTempRoot(); + writeCompatConfig(root); + const cwd = vi.spyOn(process, 'cwd').mockReturnValue(root); + + try { + const plugins = voidPlugin(); + const stubsPlugin = plugins.find((plugin) => plugin.name === 'void:client-stubs'); + const resolveId = getHookHandler(stubsPlugin?.resolveId) as ( + this: { environment?: { name?: string }; error(message: string): never }, + id: string, + ) => string | undefined; + + expect(stubsPlugin).toBeTruthy(); + expect(() => + resolveId.call( + { + environment: { name: 'client' }, + error(message: string): never { + throw new Error(message); + }, + }, + 'void/live', + ), + ).toThrow('void/live is server-only'); + } finally { + cwd.mockRestore(); + } + }); + it('includes Cloudflare builtin externalization for class A frameworks', () => { const root = createTempRoot(); writeCompatConfig(root); @@ -366,4 +421,82 @@ describe('detectFramework', () => { cwd.mockRestore(); } }); + + it('wires void/live for explicit Cloudflare targets', () => { + const root = createTempRoot(); + mkdirSync(join(root, 'routes'), { recursive: true }); + writeFileSync(join(root, 'void.json'), JSON.stringify({ target: 'cloudflare' })); + writeFileSync( + join(root, 'src', 'live.ts'), + ` + import { defineLiveStream } from "void/live"; + export const live = defineLiveStream({ id: "app" }); + `, + ); + writeFileSync( + join(root, 'routes', 'live.ts'), + ` + import { defineHandler } from "void"; + import { live } from "../src/live"; + export const GET = defineHandler((c) => live.connect(c)); + export const POST = defineHandler((c) => live.control(c)); + `, + ); + + const cwd = vi.spyOn(process, 'cwd').mockReturnValue(root); + + try { + voidPlugin(); + const entry = readFileSync(join(root, '.void', 'entry.ts'), 'utf-8'); + expect(entry).toContain('VoidLiveStreamDurableObject'); + expect(entry).toContain('export { VoidLiveStreamDurableObject };'); + expect(entry).toContain('__setLiveBindingName("VOID_LIVE");'); + } finally { + cwd.mockRestore(); + } + }); + + it('rejects void/live on non-Cloudflare targets', () => { + const root = createTempRoot(); + mkdirSync(join(root, 'routes'), { recursive: true }); + writeFileSync(join(root, 'void.json'), JSON.stringify({ target: 'node' })); + writeFileSync( + join(root, 'routes', 'live.ts'), + ` + import { defineLiveStream } from "void/live"; + export const live = defineLiveStream({ id: "app" }); + `, + ); + + const cwd = vi.spyOn(process, 'cwd').mockReturnValue(root); + + try { + expect(() => voidPlugin()).toThrow('void/live requires target: "cloudflare"'); + } finally { + cwd.mockRestore(); + } + }); + + it('rejects void/live in framework adapter mode', () => { + const root = createTempRoot(); + writeFileSync( + join(root, 'package.json'), + JSON.stringify({ dependencies: { '@sveltejs/kit': '^2.0.0' } }), + ); + writeFileSync( + join(root, 'src', 'live.ts'), + ` + import { defineLiveStream } from "void/live"; + export const live = defineLiveStream({ id: "app" }); + `, + ); + + const cwd = vi.spyOn(process, 'cwd').mockReturnValue(root); + + try { + expect(() => voidPlugin()).toThrow('void/live is currently supported in Void apps only'); + } finally { + cwd.mockRestore(); + } + }); }); diff --git a/packages/void/test/unit/wrangler.test.ts b/packages/void/test/unit/wrangler.test.ts index a8bdc2a21..e21ca104e 100644 --- a/packages/void/test/unit/wrangler.test.ts +++ b/packages/void/test/unit/wrangler.test.ts @@ -6,7 +6,9 @@ import { ensureWranglerCompatibilityDate, ensureWranglerNodejsAlsFlag, LATEST_KNOWN_COMPAT_DATE, + readExistingLiveBindingConfig, readWranglerCompat, + resolveLiveBindingConfig, syncWranglerBindings, } from '../../src/cli/wrangler.ts'; import type { VoidConfig } from '../../src/config.ts'; @@ -414,6 +416,97 @@ describe('syncWranglerBindings', () => { ]); }); + it('adds live durable object and SQLite migration metadata', () => { + setup('wrangler.jsonc', '{ "name": "test" }'); + const result = syncWranglerBindings(tmpDir, { ...noBindings, needsLive: true }); + expect(result).toBe(true); + const parsed = JSON.parse(readFileSync(join(tmpDir, 'wrangler.jsonc'), 'utf-8')); + expect(parsed.durable_objects.bindings).toEqual([ + { name: 'VOID_LIVE', class_name: 'VoidLiveStreamDurableObject' }, + ]); + expect(parsed.migrations).toEqual([ + { tag: 'live-v1', new_sqlite_classes: ['VoidLiveStreamDurableObject'] }, + ]); + }); + + it('does not duplicate live durable object bindings or migrations', () => { + setup( + 'wrangler.jsonc', + JSON.stringify({ + durable_objects: { + bindings: [{ name: 'VOID_LIVE', class_name: 'VoidLiveStreamDurableObject' }], + }, + migrations: [{ tag: 'live-v1', new_sqlite_classes: ['VoidLiveStreamDurableObject'] }], + }), + ); + const result = syncWranglerBindings(tmpDir, { ...noBindings, needsLive: true }); + expect(result).toBe(false); + expect(readExistingLiveBindingConfig(tmpDir)).toEqual({ + bindingName: 'VOID_LIVE', + className: 'VoidLiveStreamDurableObject', + }); + }); + + it('adds a non-conflicting live migration when live-v1 exists for another class', () => { + setup( + 'wrangler.jsonc', + JSON.stringify({ + migrations: [{ tag: 'live-v1', new_sqlite_classes: ['OtherDurableObject'] }], + }), + ); + const result = syncWranglerBindings(tmpDir, { ...noBindings, needsLive: true }); + expect(result).toBe(true); + const parsed = JSON.parse(readFileSync(join(tmpDir, 'wrangler.jsonc'), 'utf-8')); + expect(parsed.migrations).toEqual([ + { tag: 'live-v1', new_sqlite_classes: ['OtherDurableObject'] }, + { tag: 'live-v1-2', new_sqlite_classes: ['VoidLiveStreamDurableObject'] }, + ]); + }); + + it('uses deterministic live names when defaults collide', () => { + setup( + 'wrangler.jsonc', + JSON.stringify({ + durable_objects: { + bindings: [{ name: 'VOID_LIVE', class_name: 'UserLiveDurableObject' }], + }, + }), + ); + const live = resolveLiveBindingConfig(tmpDir); + const result = syncWranglerBindings(tmpDir, { ...noBindings, needsLive: true }, { live }); + expect(result).toBe(true); + expect(live.bindingName).toMatch(/^VOID_LIVE_[0-9a-f]{8}$/); + expect(live.className).toBe('VoidLiveStreamDurableObject'); + const parsed = JSON.parse(readFileSync(join(tmpDir, 'wrangler.jsonc'), 'utf-8')); + expect(parsed.durable_objects.bindings).toEqual([ + { name: 'VOID_LIVE', class_name: 'UserLiveDurableObject' }, + { name: live.bindingName, class_name: live.className }, + ]); + expect(parsed.migrations).toEqual([{ tag: 'live-v1', new_sqlite_classes: [live.className] }]); + }); + + it('does not treat arbitrary live-like prefixes as generated live names', () => { + setup( + 'wrangler.jsonc', + JSON.stringify({ + durable_objects: { + bindings: [ + { + name: 'VOID_LIVE_custom', + class_name: 'VoidLiveStreamDurableObject_custom', + }, + ], + }, + }), + ); + + expect(resolveLiveBindingConfig(tmpDir)).toEqual({ + bindingName: 'VOID_LIVE', + className: 'VoidLiveStreamDurableObject', + }); + expect(readExistingLiveBindingConfig(tmpDir)).toBeUndefined(); + }); + it('uses custom sandbox config', () => { setup('wrangler.jsonc', '{ "name": "test" }'); syncWranglerBindings( diff --git a/packages/void/vite.config.ts b/packages/void/vite.config.ts index efc2d640a..3efda1177 100644 --- a/packages/void/vite.config.ts +++ b/packages/void/vite.config.ts @@ -40,6 +40,8 @@ function injectWorkersTypesReferencePlugin() { 'runtime/handler.d.mts', 'runtime/env-public.d.mts', 'runtime/sandbox.d.mts', + 'runtime/live.d.mts', + 'runtime/live-server.d.mts', ]); if (!targets.has(output.fileName)) { continue; @@ -77,6 +79,8 @@ export default defineConfig({ 'src/runtime/fetch-stream.ts', 'src/runtime/sse.ts', 'src/runtime/sse-client.ts', + 'src/runtime/live.ts', + 'src/runtime/live-client.ts', 'src/runtime/auth.ts', 'src/runtime/auth-client.ts', 'src/runtime/ws.ts', @@ -110,6 +114,7 @@ export default defineConfig({ 'src/runtime/migration-handler.ts', 'src/runtime/migration-handler-pg.ts', 'src/runtime/ws-server.ts', + 'src/runtime/live-server.ts', 'src/sveltekit.ts', 'src/pages/index.ts', 'src/pages/protocol.ts', diff --git a/playground/react-pages/lib/live.ts b/playground/react-pages/lib/live.ts new file mode 100644 index 000000000..bd7c10e3a --- /dev/null +++ b/playground/react-pages/lib/live.ts @@ -0,0 +1,8 @@ +import { defineLiveStream } from 'void/live'; + +export const live = defineLiveStream({ + id: 'react-pages-playground', + allowAnonymousControl: true, +}); + +export const LIVE_PLAYGROUND_TOPIC = 'playground:react-pages'; diff --git a/playground/react-pages/pages/index.tsx b/playground/react-pages/pages/index.tsx index 8a6aea8de..8733d0d83 100644 --- a/playground/react-pages/pages/index.tsx +++ b/playground/react-pages/pages/index.tsx @@ -10,7 +10,7 @@ export default function HomePage({ title }: Props) { Go to Users {' '} Link behavior{' '} - Form behavior lab + Form behavior lab Live streams ); } diff --git a/playground/react-pages/pages/layout.tsx b/playground/react-pages/pages/layout.tsx index 42460f129..c40a16e15 100644 --- a/playground/react-pages/pages/layout.tsx +++ b/playground/react-pages/pages/layout.tsx @@ -21,6 +21,10 @@ export default function Layout({ children }: { children: React.ReactNode }) { Deferred {' '} |{' '} + + Live + {' '} + |{' '} Form Lab {' '} diff --git a/playground/react-pages/pages/live.server.ts b/playground/react-pages/pages/live.server.ts new file mode 100644 index 000000000..98317575e --- /dev/null +++ b/playground/react-pages/pages/live.server.ts @@ -0,0 +1,17 @@ +import { defineHandler, defineHead } from 'void'; +import { LIVE_PLAYGROUND_TOPIC } from '../lib/live.ts'; + +export interface Props { + topic: string; +} + +export const loader = defineHandler(() => { + return { topic: LIVE_PLAYGROUND_TOPIC }; +}); + +export const head = defineHead(() => { + return { + title: 'Live Streams', + meta: [{ name: 'description', content: 'Live stream playground for React Pages' }], + }; +}); diff --git a/playground/react-pages/pages/live.tsx b/playground/react-pages/pages/live.tsx new file mode 100644 index 000000000..04ba9f9c2 --- /dev/null +++ b/playground/react-pages/pages/live.tsx @@ -0,0 +1,140 @@ +import { useEffect, useMemo, useState } from 'react'; +import type { LiveClientEvent } from 'void/live'; +import { connectLiveStream } from 'void/live/client'; +import type { Props } from './live.server.ts'; + +type PlaygroundEvent = { + text: string; + sentAt: string; +}; + +type Status = 'connecting' | 'connected' | 'closed' | 'error'; + +export default function LivePage({ topic }: Props) { + const [status, setStatus] = useState('connecting'); + const [message, setMessage] = useState('Hello from the browser'); + const [events, setEvents] = useState>>([]); + const [error, setError] = useState(null); + const [pending, setPending] = useState(false); + const endpoint = '/api/live'; + + useEffect(() => { + const stream = connectLiveStream(endpoint, { + retryDelay: 500, + onError(errorEvent) { + setStatus('error'); + setError(errorEvent instanceof Error ? errorEvent.message : 'Live stream error'); + }, + }); + let cancelled = false; + let unsubscribe: (() => Promise) | undefined; + + stream + .subscribe({ + id: 'react-pages-live-demo', + topic, + onEvent(event) { + setEvents((current) => [event, ...current].slice(0, 12)); + setStatus('connected'); + setError(null); + }, + }) + .then((fn) => { + if (cancelled) { + void fn(); + return; + } + unsubscribe = fn; + setStatus('connected'); + }) + .catch((err: unknown) => { + if (cancelled) { + return; + } + setStatus('error'); + setError(err instanceof Error ? err.message : String(err)); + }); + + return () => { + cancelled = true; + setStatus('closed'); + void unsubscribe?.(); + stream.close(); + }; + }, [topic]); + + const statusLabel = useMemo(() => { + switch (status) { + case 'connecting': + return 'Connecting'; + case 'connected': + return 'Connected'; + case 'closed': + return 'Closed'; + case 'error': + return 'Error'; + } + }, [status]); + + async function publish() { + setPending(true); + setError(null); + try { + const response = await fetch('/api/live/publish', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ message }), + }); + if (!response.ok) { + throw new Error(await response.text()); + } + setMessage(''); + } catch (err) { + setStatus('error'); + setError(err instanceof Error ? err.message : String(err)); + } finally { + setPending(false); + } + } + + return ( +
+

Live Streams

+

+ Topic: {topic} +

+

Status: {statusLabel}

+ {error &&

{error}

} + +
{ + event.preventDefault(); + void publish(); + }} + > + {' '} + +
+ +

Events

+ {events.length === 0 ? ( +

No events yet. Publish a message to see it fan out over SSE.

+ ) : ( +
    + {events.map((event, index) => ( +
  1. + {event.data.text}{' '} + + {event.type ?? 'message'} · {event.eventId ?? 'no event id'} · {event.data.sentAt} + +
    2. + ))} +
+ )} +
+ ); +} diff --git a/playground/react-pages/routes/api/live.ts b/playground/react-pages/routes/api/live.ts new file mode 100644 index 000000000..a693bdde7 --- /dev/null +++ b/playground/react-pages/routes/api/live.ts @@ -0,0 +1,6 @@ +import { defineHandler } from 'void'; +import { live } from '../../lib/live.ts'; + +export const GET = defineHandler((c) => live.connect(c)); + +export const POST = defineHandler((c) => live.control(c)); diff --git a/playground/react-pages/routes/api/live/publish.ts b/playground/react-pages/routes/api/live/publish.ts new file mode 100644 index 000000000..dcdf6d99a --- /dev/null +++ b/playground/react-pages/routes/api/live/publish.ts @@ -0,0 +1,25 @@ +import { defineHandler } from 'void'; +import { LIVE_PLAYGROUND_TOPIC, live } from '../../../lib/live.ts'; + +export const POST = defineHandler(async (c) => { + const body = await c.req.json<{ message?: unknown }>().catch(() => null); + const text = + typeof body?.message === 'string' && body.message.trim() + ? body.message.trim() + : 'Hello from the React Pages playground'; + const eventId = crypto.randomUUID(); + + await live.publish( + LIVE_PLAYGROUND_TOPIC, + { + text, + sentAt: new Date().toISOString(), + }, + { + type: 'message', + eventId, + }, + ); + + return c.json({ ok: true, eventId }); +});