chore(cloudflare-workers): upgrade to new manager api (#1253)

Author: NathanFlurry

examples/cloudflare-workers/scripts/client.ts

@@ -3,7 +3,7 @@ import type { registry } from "../src/registry";
 
 // Create RivetKit client
 const client = createClient<typeof registry>(
-	process.env.RIVETKIT_ENDPOINT ?? "http://localhost:8787",
+	process.env.RIVETKIT_ENDPOINT ?? "http://localhost:8787/rivet",
 );
 
 async function main() {
@@ -12,8 +12,6 @@ async function main() {
 	try {
 		// Create counter instance
 		const counter = client.counter.getOrCreate("demo");
-		const conn = counter.connect();
-		conn.on("foo", (x) => console.log("output", x));
 
 		// Increment counter
 		console.log("Incrementing counter 'demo'...");
@@ -36,6 +34,33 @@ async function main() {
 		console.error("❌ Error:", error);
 		process.exit(1);
 	}
+	// try {
+	// 	// Create counter instance
+	// 	const counter = client.counter.getOrCreate("demo");
+	// 	const conn = counter.connect();
+	// 	conn.on("foo", (x) => console.log("output", x));
+	//
+	// 	// Increment counter
+	// 	console.log("Incrementing counter 'demo'...");
+	// 	const result1 = await counter.increment(1);
+	// 	console.log("New count:", result1);
+	//
+	// 	// Increment again with larger value
+	// 	console.log("Incrementing counter 'demo' by 5...");
+	// 	const result2 = await counter.increment(5);
+	// 	console.log("New count:", result2);
+	//
+	// 	// Create another counter
+	// 	const counter2 = client.counter.getOrCreate("another");
+	// 	console.log("Incrementing counter 'another' by 10...");
+	// 	const result3 = await counter2.increment(10);
+	// 	console.log("New count:", result3);
+	//
+	// 	console.log("✅ Demo completed!");
+	// } catch (error) {
+	// 	console.error("❌ Error:", error);
+	// 	process.exit(1);
+	// }
 }
 
 main().catch(console.error);

examples/cloudflare-workers/src/index.ts

@@ -1,5 +1,5 @@
-import { createServerHandler } from "@rivetkit/cloudflare-workers";
+import { createHandler } from "@rivetkit/cloudflare-workers";
 import { registry } from "./registry";
 
-const { handler, ActorHandler } = createServerHandler(registry);
+const { handler, ActorHandler } = createHandler(registry);
 export { handler as default, ActorHandler };

packages/cloudflare-workers/src/config.ts

@@ -5,7 +5,10 @@ import { z } from "zod";
 export const ConfigSchema = RunConfigSchema.removeDefault()
 	.omit({ driver: true, getUpgradeWebSocket: true })
 	.extend({
-		app: z.custom<Hono>().optional(),
+		/** Path that the Rivet manager API will be mounted. */
+		managerPath: z.string().optional().default("/rivet"),
+
+		fetch: z.custom<ExportedHandlerFetchHandler<unknown, unknown>>().optional(),
 	})
 	.default({});
 export type InputConfig = z.input<typeof ConfigSchema>;

packages/cloudflare-workers/src/handler.ts

@@ -1,5 +1,5 @@
 import { env } from "cloudflare:workers";
-import { Hono } from "hono";
+import type { Hono } from "hono";
 import type { Registry, RunConfig } from "rivetkit";
 import type { Client } from "rivetkit/client";
 import {
@@ -36,7 +36,7 @@ interface SetupOutput<A extends Registry<any>> {
 	createHandler: (hono?: Hono) => Handler;
 }
 
-export function createServerHandler<R extends Registry<any>>(
+export function createHandler<R extends Registry<any>>(
 	registry: R,
 	inputConfig?: InputConfig,
 ): Handler {
@@ -66,23 +66,34 @@ export function createServer<R extends Registry<any>>(
 	const ActorHandler = createActorDurableObject(registry, runConfig);
 
 	// Create server
-	const serverOutput = registry.createServer(runConfig);
+	const serverOutput = registry.start(runConfig);
 
 	return {
 		client: serverOutput.client as Client<R>,
-		createHandler: (hono) => {
-			// Build base router
-			const app = hono ?? new Hono();
-
-			// Mount registry router at /registry
-			if (!hono) {
-				app.route("/registry", serverOutput.hono);
-			}
-
+		createHandler: () => {
 			// Create Cloudflare handler
 			const handler = {
 				fetch: (request, env, ctx) => {
-					return app.fetch(request, env, ctx);
+					const url = new URL(request.url);
+
+					// Mount Rivet manager API
+					if (url.pathname.startsWith(config.managerPath)) {
+						const strippedPath = url.pathname.substring(
+							config.managerPath.length,
+						);
+						url.pathname = strippedPath;
+						const modifiedRequest = new Request(url.toString(), request);
+						return serverOutput.fetch(modifiedRequest, env, ctx);
+					}
+
+					if (config.fetch) {
+						return config.fetch(request, env, ctx);
+					} else {
+						return new Response(
+							"This is a RivetKit server.\n\nLearn more at https://rivetkit.org\n",
+							{ status: 200 },
+						);
+					}
 				},
 			} satisfies ExportedHandler<Bindings>;
 

packages/cloudflare-workers/src/mod.ts

@@ -1,3 +1,3 @@
 export type { DriverContext } from "./actor-driver";
 export type { InputConfig as Config } from "./config";
-export { type Bindings, createServer, createServerHandler } from "./handler";
+export { type Bindings, createHandler, createServer } from "./handler";

packages/cloudflare-workers/tests/driver-tests.test.ts

@@ -141,15 +141,12 @@ async function setupProject(projectPath: string) {
 	await fs.mkdir(rivetKitDir, { recursive: true });
 
 	// Copy core package
-	const corePackagePath = path.join(workspaceRoot, "packages/core");
-	const targetCorePath = path.join(rivetKitDir, "core");
+	const corePackagePath = path.join(workspaceRoot, "packages/rivetkit");
+	const targetCorePath = path.join(rivetKitDir, "rivetkit");
 	await fs.cp(corePackagePath, targetCorePath, { recursive: true });
 
 	// Copy cloudflare-workers package
-	const cfPackagePath = path.join(
-		workspaceRoot,
-		"packages/drivers/cloudflare-workers",
-	);
+	const cfPackagePath = path.join(workspaceRoot, "packages/cloudflare-workers");
 	const targetCfPath = path.join(rivetKitDir, "cloudflare-workers");
 	await fs.cp(cfPackagePath, targetCfPath, { recursive: true });
 

packages/rivetkit/src/actor/errors.ts

@@ -165,7 +165,7 @@ export class MessageTooLong extends ActorError {
 		super(
 			"message",
 			"too_long",
-			"Message too long. This can be configured with: `registry.runServer({ maxIncomingMessageSize: ... })`",
+			"Message too long. This can be configured with: `registry.start({ maxIncomingMessageSize: ... })`",
 			{ public: true },
 		);
 	}

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

@@ -21,7 +21,7 @@ export async function setupDriverTest(
 	}
 
 	// Build drivers
-	const projectPath = resolve(__dirname, "../../fixtures/driver-test-suite");
+	const projectPath = resolve(__dirname, "../../../fixtures/driver-test-suite");
 	const { endpoint, namespace, runnerName, cleanup } =
 		await driverTestConfig.start(projectPath);
 	c.onTestFinished(cleanup);

packages/rivetkit/src/registry/mod.ts

@@ -21,12 +21,10 @@ import {
 import { crossPlatformServe } from "./serve";
 
 interface ServerOutput<A extends Registry<any>> {
-	config: RunConfig;
-	driver: DriverConfig;
+	/** Client to communicate with the actors. */
 	client: Client<A>;
-	hono: Hono;
-	handler: (req: Request) => Promise<Response>;
-	serve: (hono?: Hono) => void;
+	/** Fetch handler to manually route requests to the Rivet manager API. */
+	fetch: (request: Request, ...args: any) => Response | Promise<Response>;
 }
 
 export class Registry<A extends RegistryActors> {
@@ -43,7 +41,7 @@ export class Registry<A extends RegistryActors> {
 	/**
 	 * Runs the registry for a server.
 	 */
-	public createServer(inputConfig?: RunConfigInput): ServerOutput<this> {
+	public start(inputConfig?: RunConfigInput): ServerOutput<this> {
 		const config = RunConfigSchema.parse(inputConfig);
 
 		// Configure logger
@@ -62,6 +60,13 @@ export class Registry<A extends RegistryActors> {
 		// TODO: Find cleaner way of disabling by default
 		if (driver.name === "engine") {
 			config.inspector.enabled = false;
+			config.disableServer = true;
+		}
+		if (driver.name === "cloudflare-workers") {
+			config.inspector.enabled = false;
+			config.disableServer = true;
+			config.disableActorDriver = true;
+			config.noWelcome = true;
 		}
 
 		// Configure getUpgradeWebSocket lazily so we can assign it in crossPlatformServe
@@ -109,41 +114,32 @@ export class Registry<A extends RegistryActors> {
 			console.log();
 		}
 
+		// 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
-		const _actorDriver = driver.actor(
-			this.#config,
-			config,
-			managerDriver,
-			client,
-		);
+		if (!config.disableActorDriver) {
+			const _actorDriver = driver.actor(
+				this.#config,
+				config,
+				managerDriver,
+				client,
+			);
+		}
+
+		// Start server
+		if (!config.disableServer) {
+			(async () => {
+				const out = await crossPlatformServe(hono, undefined);
+				upgradeWebSocket = out.upgradeWebSocket;
+			})();
+		}
 
 		return {
-			config,
-			driver,
 			client,
-			hono,
-			handler: async (req: Request) => await hono.fetch(req),
-			serve: async (app) => {
-				const out = await crossPlatformServe(hono, app);
-				upgradeWebSocket = out.upgradeWebSocket;
-			},
+			fetch: hono.fetch.bind(hono),
 		};
 	}
-
-	/**
-	 * Runs the registry as a standalone server.
-	 */
-	public async runServer(inputConfig?: RunConfigInput) {
-		const { driver, serve } = this.createServer(inputConfig);
-
-		// TODO: FInd better way of doing this
-		// Don't run server by default
-		if (driver.name !== "engine") {
-			serve();
-		}
-	}
 }
 
 export function setup<A extends RegistryActors>(

packages/rivetkit/src/registry/run-config.ts

@@ -29,16 +29,31 @@ export const RunConfigSchema = ClientConfigSchema.extend({
 
 	inspector: InspectorConfigSchema,
 
+	/** @experimental */
+	disableServer: z.boolean().optional().default(false),
+
+	/** @experimental */
+	disableActorDriver: z.boolean().optional().default(false),
+
 	/**
+	 * @experimental
+	 *
 	 * Base path for the router. This is used to prefix all routes.
 	 * For example, if the base path is `/api`, then the route `/actors` will be
 	 * available at `/api/actors`.
 	 */
 	basePath: z.string().optional().default("/"),
 
-	/** Disable welcome message. */
+	/**
+	 * @experimental
+	 *
+	 * Disable welcome message.
+	 * */
 	noWelcome: z.boolean().optional().default(false),
 
+	/**
+	 * @experimental
+	 * */
 	logging: z
 		.object({
 			baseLogger: z.custom<Logger>().optional(),