feat: serverless

Author: MasterPtato

examples/next-js/src/app/api/registry/[...all]/route.ts renamed to examples/next-js/src/app/api/rivet/[...all]/route.ts

@@ -1,4 +1,4 @@
 import { toNextHandler } from "@rivetkit/next-js";
 import { registry } from "@/rivet/registry";
 
-export const { GET, POST, HEAD, PATCH, OPTIONS } = toNextHandler(registry);
+export const { GET, POST, PUT, PATCH, HEAD, OPTIONS } = toNextHandler(registry);

packages/next-js/src/mod.ts

@@ -9,11 +9,22 @@ export const toNextHandler = (
 
 	const { fetch } = registry.start(inputConfig);
 
+	const fetchWrapper = async (request: Request, { params }: { params: Promise<{ all: string[] }> }) => {
+		const { all } = await params;
+
+		const newUrl = new URL(request.url);
+		newUrl.pathname = all.join('/');
+		const newReq = new Request(newUrl, request);
+
+		return await fetch(newReq);
+	};
+
 	return {
-		GET: fetch,
-		POST: fetch,
-		PATCH: fetch,
-		HEAD: fetch,
-		OPTIONS: fetch,
+		GET: fetchWrapper,
+		POST: fetchWrapper,
+		PUT: fetchWrapper,
+		PATCH: fetchWrapper,
+		HEAD: fetchWrapper,
+		OPTIONS: fetchWrapper,
 	};
 };

packages/rivetkit/src/actor/driver.ts

@@ -7,6 +7,7 @@ import type * as protocol from "@/schemas/client-protocol/mod";
 import type { AnyConn, ConnectionDriver } from "./connection";
 import type { GenericConnGlobalState } from "./generic-conn-driver";
 import type { AnyActorInstance } from "./instance";
+import { Hono, MiddlewareHandler } from "hono";
 
 export type ConnectionDriversMap = Record<ConnectionDriver, ConnDriver>;
 
@@ -45,6 +46,8 @@ export interface ActorDriver {
 	sleep?(actorId: string): Promise<void>;
 
 	shutdown?(immediate: boolean): Promise<void>;
+
+	modifyManagerRouter?: (router: Hono, cors: MiddlewareHandler) => void;
 }
 
 export enum ConnectionReadyState {

packages/rivetkit/src/common/versioned-data.ts

@@ -13,7 +13,7 @@ export interface VersionedDataConfig<T> {
 }
 
 export class VersionedDataHandler<T> {
-	constructor(private config: VersionedDataConfig<T>) {}
+	constructor(private config: VersionedDataConfig<T>) { }
 
 	serializeWithEmbeddedVersion(data: T): Uint8Array {
 		const versioned: VersionedData<Uint8Array> = {
@@ -63,8 +63,8 @@ export class VersionedDataHandler<T> {
 	}
 
 	private embedVersion(data: VersionedData<Uint8Array>): Uint8Array {
-		const versionBytes = new Uint8Array(4);
-		new DataView(versionBytes.buffer).setUint32(0, data.version, true);
+		const versionBytes = new Uint8Array(2);
+		new DataView(versionBytes.buffer).setUint16(0, data.version, true);
 
 		const result = new Uint8Array(versionBytes.length + data.data.length);
 		result.set(versionBytes);
@@ -74,15 +74,15 @@ export class VersionedDataHandler<T> {
 	}
 
 	private extractVersion(bytes: Uint8Array): VersionedData<Uint8Array> {
-		if (bytes.length < 4) {
+		if (bytes.length < 2) {
 			throw new Error("Invalid versioned data: too short");
 		}
 
-		const version = new DataView(bytes.buffer, bytes.byteOffset).getUint32(
+		const version = new DataView(bytes.buffer, bytes.byteOffset).getUint16(
 			0,
 			true,
 		);
-		const data = bytes.slice(4);
+		const data = bytes.slice(2);
 
 		return { version, data };
 	}

packages/rivetkit/src/driver-test-suite/mod.ts

@@ -197,6 +197,7 @@ export async function createTestRuntime(
 			registry.config,
 			config,
 			managerDriver,
+			undefined,
 			false,
 		);
 

packages/rivetkit/src/drivers/engine/actor-driver.ts

@@ -40,6 +40,8 @@ import type { RunConfig } from "@/registry/run-config";
 import type { Config } from "./config";
 import { KEYS } from "./kv";
 import { logger } from "./log";
+import { Hono, MiddlewareHandler } from "hono";
+import { streamSSE } from "hono/streaming";
 
 interface ActorHandler {
 	actor?: AnyActorInstance;
@@ -61,6 +63,9 @@ export class EngineActorDriver implements ActorDriver {
 	#actorRouter: ActorRouter;
 	#version: number = 1; // Version for the runner protocol
 
+	#runnerStarted: PromiseWithResolvers<undefined> = Promise.withResolvers();
+	#runnerStopped: PromiseWithResolvers<undefined> = Promise.withResolvers();
+
 	constructor(
 		registryConfig: RegistryConfig,
 		runConfig: RunConfig,
@@ -108,6 +113,8 @@ export class EngineActorDriver implements ActorDriver {
 						runnerName: this.#config.runnerName,
 					});
 				}
+
+				this.#runnerStarted.resolve(undefined);
 			},
 			onDisconnected: () => {
 				logger().warn({
@@ -117,7 +124,9 @@ export class EngineActorDriver implements ActorDriver {
 				});
 				hasDisconnected = true;
 			},
-			onShutdown: () => {},
+			onShutdown: () => {
+				this.#runnerStopped.resolve(undefined);
+			},
 			fetch: this.#runnerFetch.bind(this),
 			websocket: this.#runnerWebSocket.bind(this),
 			onActorStart: this.#runnerOnActorStart.bind(this),
@@ -367,4 +376,22 @@ export class EngineActorDriver implements ActorDriver {
 		logger().info({ msg: "stopping engine actor driver" });
 		await this.#runner.shutdown(immediate);
 	}
+
+	modifyManagerRouter(router: Hono, cors: MiddlewareHandler) {
+		// Serverless start endpoint
+		router.get("/start", cors, async (c) => {
+			await this.#runnerStarted.promise;
+
+			return streamSSE(c, async (stream) => {
+				// Runner id should be set if the runner started
+				stream.writeSSE({ data: this.#runner.runnerId! });
+
+				await this.#runnerStopped.promise;
+			});
+		});
+
+		router.get("/health", cors, (c) => {
+			return c.text("ok");
+		});
+	}
 }

packages/rivetkit/src/drivers/file-system/actor.ts

@@ -9,6 +9,8 @@ import type {
 import type { RegistryConfig, RunConfig } from "@/mod";
 import { bufferToArrayBuffer } from "@/utils";
 import type { FileSystemGlobalState } from "./global-state";
+import { Hono, MiddlewareHandler } from "hono";
+import { streamSSE } from "hono/streaming";
 
 export type ActorDriverContext = Record<never, never>;
 

packages/rivetkit/src/manager/router.ts

@@ -20,7 +20,7 @@ import {
 	loggerMiddleware,
 } from "@/common/router";
 import { deconstructError, noopNext } from "@/common/utils";
-import { HEADER_ACTOR_ID } from "@/driver-helpers/mod";
+import { ActorDriver, HEADER_ACTOR_ID } from "@/driver-helpers/mod";
 import type {
 	TestInlineDriverCallRequest,
 	TestInlineDriverCallResponse,
@@ -54,10 +54,10 @@ function buildOpenApiResponses<T>(schema: T, validateBody: boolean) {
 			description: "Success",
 			content: validateBody
 				? {
-						"application/json": {
-							schema,
-						},
-					}
+					"application/json": {
+						schema,
+					},
+				}
 				: {},
 		},
 		400: {
@@ -73,6 +73,7 @@ export function createManagerRouter(
 	registryConfig: RegistryConfig,
 	runConfig: RunConfig,
 	managerDriver: ManagerDriver,
+	actorDriver: ActorDriver | undefined,
 	validateBody: boolean,
 ): { router: Hono; openapi: OpenAPIHono } {
 	const router = new OpenAPIHono({ strict: false }).basePath(
@@ -151,6 +152,7 @@ export function createManagerRouter(
 				method: c.req.raw.method,
 				headers: proxyHeaders,
 				body: c.req.raw.body,
+				duplex: "half",
 				signal: c.req.raw.signal,
 			});
 
@@ -211,10 +213,10 @@ export function createManagerRouter(
 				body: {
 					content: validateBody
 						? {
-								"application/json": {
-									schema: ActorsGetOrCreateByIdRequestSchema,
-								},
-							}
+							"application/json": {
+								schema: ActorsGetOrCreateByIdRequestSchema,
+							},
+						}
 						: {},
 				},
 			},
@@ -323,10 +325,10 @@ export function createManagerRouter(
 				body: {
 					content: validateBody
 						? {
-								"application/json": {
-									schema: ActorsCreateRequestSchema,
-								},
-							}
+							"application/json": {
+								schema: ActorsCreateRequestSchema,
+							},
+						}
 						: {},
 				},
 			},
@@ -500,6 +502,7 @@ export function createManagerRouter(
 						method: c.req.method,
 						headers: c.req.raw.headers,
 						body: c.req.raw.body,
+						duplex: "half",
 					}),
 				);
 
@@ -532,6 +535,8 @@ export function createManagerRouter(
 		router as unknown as Hono,
 	);
 
+	actorDriver?.modifyManagerRouter?.(router as unknown as Hono, cors);
+
 	if (runConfig.inspector?.enabled) {
 		if (!managerDriver.inspector) {
 			throw new Unsupported("inspector");
@@ -598,9 +603,9 @@ async function createTestWebSocketProxy(
 			onOpen: (_evt, serverWs) => {
 				serverWs.close(1011, "Failed to establish connection");
 			},
-			onMessage: () => {},
-			onError: () => {},
-			onClose: () => {},
+			onMessage: () => { },
+			onError: () => { },
+			onClose: () => { },
 		};
 	}
 

packages/rivetkit/src/registry/mod.ts

@@ -76,12 +76,6 @@ export class Registry<A extends RegistryActors> {
 
 		// Create router
 		const managerDriver = driver.manager(this.#config, config);
-		const { router: hono } = createManagerRouter(
-			this.#config,
-			config,
-			managerDriver,
-			false,
-		);
 
 		// Create client
 		const client = createClientWithDriver<this>(managerDriver, config);
@@ -115,17 +109,24 @@ export class Registry<A extends RegistryActors> {
 
 		// HACK: We need to find a better way to let the driver itself decide when to start the actor driver
 		// Create runner
-		//
-		// Even though we do not use the return value, this is required to start the code that will handle incoming actors
+		let actorDriver;
 		if (!config.disableActorDriver) {
-			const _actorDriver = driver.actor(
+			actorDriver = driver.actor(
 				this.#config,
 				config,
 				managerDriver,
 				client,
 			);
 		}
 
+		const { router: hono } = createManagerRouter(
+			this.#config,
+			config,
+			managerDriver,
+			actorDriver,
+			false,
+		);
+
 		// Start server
 		if (!config.disableServer) {
 			(async () => {

packages/rivetkit/src/remote-manager-driver/actor-http-client.ts

@@ -17,7 +17,6 @@ export async function sendHttpRequestToActor(
 	const guardHeaders = buildGuardHeadersForHttp(actorRequest, actorId);
 
 	if (
-		actorRequest.body &&
 		actorRequest.method !== "GET" &&
 		actorRequest.method !== "HEAD"
 	) {
@@ -27,17 +26,19 @@ export async function sendHttpRequestToActor(
 
 		// TODO: This buffers the entire request in memory every time. We
 		// need to properly implement streaming bodies.
-		// Clone and read the body to ensure it can be sent
-		const clonedRequest = actorRequest.clone();
-		bodyToSend = await clonedRequest.arrayBuffer();
+		const reqBody = await actorRequest.arrayBuffer();
 
-		// If this is a streaming request, we need to convert the headers
-		// for the basic array buffer
-		guardHeaders.delete("transfer-encoding");
-		guardHeaders.set(
-			"content-length",
-			String((bodyToSend as ArrayBuffer).byteLength),
-		);
+		if (reqBody.byteLength != 0) {
+			bodyToSend = reqBody;
+
+			// If this is a streaming request, we need to convert the headers
+			// for the basic array buffer
+			guardHeaders.delete("transfer-encoding");
+			guardHeaders.set(
+				"content-length",
+				String(bodyToSend.byteLength),
+			);
+		}
 	}
 
 	const guardRequest = new Request(guardUrl, {