feat: add inline client

packages/actor/scripts/dump-openapi.ts

@@ -41,7 +41,7 @@ function main() {
 	};
 
 	const managerRouter = createManagerRouter(appConfig, driverConfig, {
-		proxyMode: {
+		routingHandler: {
 			inline: {
 				handlers: sharedConnectionHandlers,
 			},

packages/actor/src/actor/conn-routing-handler.ts

@@ -0,0 +1,53 @@
+import type { ConnectionHandlers as ConnHandlers } from "./router-endpoints";
+import type { Context as HonoContext, HonoRequest } from "hono";
+
+/**
+ * Deterines how requests to actors should be routed.
+ *
+ * Inline handlers calls the connection handlers directly.
+ *
+ * Custom will let a custom function handle the request. This usually will proxy the request to another location.
+ */
+export type ConnRoutingHandler =
+	| {
+			inline: {
+				handlers: ConnHandlers;
+			};
+	  }
+	| {
+			custom: ConnRoutingHandlerCustom;
+	  };
+
+export interface ConnRoutingHandlerCustom {
+	sendRequest: SendRequestHandler;
+	openWebSocket: OpenWebSocketHandler;
+	proxyRequest: ProxyRequestHandler;
+	proxyWebSocket: ProxyWebSocketHandler;
+}
+
+export type BuildProxyEndpoint = (c: HonoContext, actorId: string) => string;
+
+export type SendRequestHandler = (
+	actorId: string,
+	meta: unknown | undefined,
+	actorRequest: Request,
+) => Promise<Response>;
+
+export type OpenWebSocketHandler = (
+	actorId: string,
+	meta?: unknown,
+) => Promise<WebSocket>;
+
+export type ProxyRequestHandler = (
+	c: HonoContext,
+	actorRequest: Request,
+	actorId: string,
+	meta?: unknown,
+) => Promise<Response>;
+
+export type ProxyWebSocketHandler = (
+	c: HonoContext,
+	path: string,
+	actorId: string,
+	meta?: unknown,
+) => Promise<Response>;

packages/actor/src/actor/router-endpoints.ts

@@ -21,7 +21,7 @@ import type { DriverConfig } from "@/driver-helpers/config";
 import invariant from "invariant";
 
 export interface ConnectWebSocketOpts {
-	req: HonoRequest;
+	req?: HonoRequest;
 	encoding: Encoding;
 	params: unknown;
 	actorId: string;
@@ -34,7 +34,7 @@ export interface ConnectWebSocketOutput {
 }
 
 export interface ConnectSseOpts {
-	req: HonoRequest;
+	req?: HonoRequest;
 	encoding: Encoding;
 	params: unknown;
 	actorId: string;
@@ -46,7 +46,7 @@ export interface ConnectSseOutput {
 }
 
 export interface ActionOpts {
-	req: HonoRequest;
+	req?: HonoRequest;
 	params: unknown;
 	actionName: string;
 	actionArgs: unknown[];
@@ -58,7 +58,7 @@ export interface ActionOutput {
 }
 
 export interface ConnsMessageOpts {
-	req: HonoRequest;
+	req?: HonoRequest;
 	connId: string;
 	connToken: string;
 	message: messageToServer.ToServer;

packages/actor/src/app/inline-client-driver.ts

@@ -0,0 +1,212 @@
+import * as errors from "@/actor/errors";
+import * as protoHttpAction from "@/actor/protocol/http/action";
+import { logger } from "./log";
+import type { EventSource } from "eventsource";
+import type * as wsToServer from "@/actor/protocol/message/to-server";
+import { type Encoding, serialize } from "@/actor/protocol/serde";
+import {
+	HEADER_CONN_PARAMS,
+	HEADER_ENCODING,
+	type ConnectionHandlers,
+} from "@/actor/router-endpoints";
+import { HonoRequest, type Context as HonoContext, type Next } from "hono";
+import invariant from "invariant";
+import { ClientDriver } from "@/client/client";
+import { ManagerDriver } from "@/manager/driver";
+import { ActorQuery } from "@/manager/protocol/query";
+import { ConnRoutingHandler } from "@/actor/conn-routing-handler";
+import { sendHttpRequest, serializeWithEncoding } from "@/client/utils";
+import { ActionRequest, ActionResponse } from "@/actor/protocol/http/action";
+import { assertUnreachable } from "@/actor/utils";
+
+/**
+ * Client driver that calls the manager driver inline.
+ *
+ * This driver can access private resources.
+ *
+ * This driver serves a double purpose as:
+ * - Providing the client for the internal requests
+ * - Provide the driver for the manager HTTP router (see manager/router.ts)
+ */
+export function createInlineClientDriver(
+	managerDriver: ManagerDriver,
+	routingHandler: ConnRoutingHandler,
+): ClientDriver {
+	//// Lazily import the dynamic imports so we don't have to turn `createClient` in to an aysnc fn
+	//const dynamicImports = (async () => {
+	//	// Import dynamic dependencies
+	//	const [WebSocket, EventSource] = await Promise.all([
+	//		importWebSocket(),
+	//		importEventSource(),
+	//	]);
+	//	return {
+	//		WebSocket,
+	//		EventSource,
+	//	};
+	//})();
+
+	const driver: ClientDriver = {
+		action: async <Args extends Array<unknown> = unknown[], Response = unknown>(
+			req: HonoRequest | undefined,
+			actorQuery: ActorQuery,
+			encoding: Encoding,
+			params: unknown,
+			actionName: string,
+			...args: Args
+		): Promise<Response> => {
+			// Get the actor ID and meta
+			const { actorId, meta } = await queryActor(
+				req,
+				actorQuery,
+				managerDriver,
+			);
+			logger().debug("found actor for action", { actorId, meta });
+			invariant(actorId, "Missing actor ID");
+
+			// Invoke the action
+			logger().debug("handling action", { actionName, encoding });
+			if ("inline" in routingHandler) {
+				const { output } = await routingHandler.inline.handlers.onAction({
+					req,
+					params,
+					actionName,
+					actionArgs: args,
+					actorId,
+				});
+				return output as Response;
+			} else if ("custom" in routingHandler) {
+				const responseData = await sendHttpRequest<
+					ActionRequest,
+					ActionResponse
+				>({
+					url: `http://actor/action/${encodeURIComponent(actionName)}`,
+					method: "POST",
+					headers: {
+						[HEADER_ENCODING]: encoding,
+						...(params !== undefined
+							? { [HEADER_CONN_PARAMS]: JSON.stringify(params) }
+							: {}),
+					},
+					body: { a: args } satisfies ActionRequest,
+					encoding: encoding,
+					customFetch: routingHandler.custom.sendRequest.bind(
+						undefined,
+						actorId,
+						meta,
+					),
+				});
+
+				return responseData.o as Response;
+			} else {
+				assertUnreachable(routingHandler);
+			}
+		},
+
+		resolveActorId: async (
+			req: HonoRequest | undefined,
+			actorQuery: ActorQuery,
+			_encodingKind: Encoding,
+		): Promise<string> => {
+			// Get the actor ID and meta
+			const { actorId } = await queryActor(req, actorQuery, managerDriver);
+			logger().debug("resolved actor", { actorId });
+			invariant(actorId, "missing actor ID");
+
+			return actorId;
+		},
+
+		connectWebSocket: async (
+			req: HonoRequest | undefined,
+			actorQuery: ActorQuery,
+			encodingKind: Encoding,
+		): Promise<WebSocket> => {
+			throw "UNIMPLEMENTED";
+		},
+
+		connectSse: async (
+			req: HonoRequest | undefined,
+			actorQuery: ActorQuery,
+			encodingKind: Encoding,
+			params: unknown,
+		): Promise<EventSource> => {
+			throw "UNIMPLEMENTED";
+		},
+
+		sendHttpMessage: async (
+			req: HonoRequest | undefined,
+			actorId: string,
+			encoding: Encoding,
+			connectionId: string,
+			connectionToken: string,
+			message: wsToServer.ToServer,
+		): Promise<Response> => {
+			throw "UNIMPLEMENTED";
+		},
+	};
+
+	return driver;
+}
+
+/**
+ * Query the manager driver to get or create an actor based on the provided query
+ */
+export async function queryActor(
+	req: HonoRequest | undefined,
+	query: ActorQuery,
+	driver: ManagerDriver,
+): Promise<{ actorId: string; meta?: unknown }> {
+	logger().debug("querying actor", { query });
+	let actorOutput: { actorId: string; meta?: unknown };
+	if ("getForId" in query) {
+		const output = await driver.getForId({
+			req,
+			actorId: query.getForId.actorId,
+		});
+		if (!output) throw new errors.ActorNotFound(query.getForId.actorId);
+		actorOutput = output;
+	} else if ("getForKey" in query) {
+		const existingActor = await driver.getWithKey({
+			req,
+			name: query.getForKey.name,
+			key: query.getForKey.key,
+		});
+		if (!existingActor) {
+			throw new errors.ActorNotFound(
+				`${query.getForKey.name}:${JSON.stringify(query.getForKey.key)}`,
+			);
+		}
+		actorOutput = existingActor;
+	} else if ("getOrCreateForKey" in query) {
+		const getOrCreateOutput = await driver.getOrCreateWithKey({
+			req,
+			name: query.getOrCreateForKey.name,
+			key: query.getOrCreateForKey.key,
+			input: query.getOrCreateForKey.input,
+			region: query.getOrCreateForKey.region,
+		});
+		actorOutput = {
+			actorId: getOrCreateOutput.actorId,
+			meta: getOrCreateOutput.meta,
+		};
+	} else if ("create" in query) {
+		const createOutput = await driver.createActor({
+			req,
+			name: query.create.name,
+			key: query.create.key,
+			input: query.create.input,
+			region: query.create.region,
+		});
+		actorOutput = {
+			actorId: createOutput.actorId,
+			meta: createOutput.meta,
+		};
+	} else {
+		throw new errors.InvalidRequest("Invalid query format");
+	}
+
+	logger().debug("actor query result", {
+		actorId: actorOutput.actorId,
+		meta: actorOutput.meta,
+	});
+	return { actorId: actorOutput.actorId, meta: actorOutput.meta };
+}

packages/actor/src/app/log.ts

@@ -0,0 +1,7 @@
+import { getLogger } from "@/common//log";
+
+export const LOGGER_NAME = "actor-app";
+
+export function logger() {
+	return getLogger(LOGGER_NAME);
+}

packages/actor/src/client/actor-conn.ts

@@ -251,6 +251,7 @@ enc
 
 	async #connectWebSocket() {
 		const ws = await this.#driver.connectWebSocket(
+			undefined,
 			this.#actorQuery,
 			this.#encodingKind,
 		);
@@ -281,6 +282,7 @@ enc
 
 	async #connectSse() {
 		const eventSource = await this.#driver.connectSse(
+			undefined,
 			this.#actorQuery,
 			this.#encodingKind,
 			this.#params,
@@ -649,6 +651,7 @@ enc
 				throw new errors.InternalError("Missing connection ID or token.");
 
 			const res = await this.#driver.sendHttpMessage(
+				undefined,
 				this.#actorId,
 				this.#encodingKind,
 				this.#connectionId,

packages/actor/src/client/actor-handle.ts

@@ -61,6 +61,7 @@ export class ActorHandleRaw {
 		...args: Args
 	): Promise<Response> {
 		return await this.#driver.action<Args, Response>(
+			undefined,
 			this.#actorQuery,
 			this.#encodingKind,
 			this.#params,
@@ -105,6 +106,7 @@ export class ActorHandleRaw {
 		) {
 			// TODO:
 			const actorId = await this.#driver.resolveActorId(
+				undefined,
 				this.#actorQuery,
 				this.#encodingKind,
 			);