feat: add actor proxy support for HTTP fallback

Author: NathanFlurry

packages/actor-core/src/actor/errors.ts

@@ -194,3 +194,55 @@ export class UserError extends ActorError {
 		});
 	}
 }
+
+// Proxy-related errors
+
+export class MissingRequiredParameters extends ActorError {
+	constructor(missingParams: string[]) {
+		super(
+			"missing_required_parameters", 
+			`Missing required parameters: ${missingParams.join(", ")}`,
+			{ public: true }
+		);
+	}
+}
+
+export class InvalidQueryJSON extends ActorError {
+	constructor(error?: unknown) {
+		super(
+			"invalid_query_json", 
+			`Invalid query JSON: ${error}`,
+			{ public: true, cause: error }
+		);
+	}
+}
+
+export class InvalidQueryFormat extends ActorError {
+	constructor(error?: unknown) {
+		super(
+			"invalid_query_format", 
+			`Invalid query format: ${error}`,
+			{ public: true, cause: error }
+		);
+	}
+}
+
+export class ActorNotFound extends ActorError {
+	constructor(identifier?: string) {
+		super(
+			"actor_not_found", 
+			identifier ? `Actor not found: ${identifier}` : "Actor not found",
+			{ public: true }
+		);
+	}
+}
+
+export class ProxyError extends ActorError {
+	constructor(operation: string, error?: unknown) {
+		super(
+			"proxy_error", 
+			`Error proxying ${operation}: ${error}`,
+			{ public: true, cause: error }
+		);
+	}
+}

packages/actor-core/src/client/actor_conn.ts

@@ -95,6 +95,7 @@ export class ActorConnRaw {
 		private readonly supportedTransports: Transport[],
 		private readonly serverTransports: Transport[],
 		private readonly dynamicImports: DynamicImports,
+		private readonly actorQuery: unknown,
 	) {
 		this.#keepNodeAliveInterval = setInterval(() => 60_000);
 	}
@@ -115,34 +116,135 @@ export class ActorConnRaw {
 	): Promise<Response> {
 		logger().debug("action", { name, args });
 
-		// TODO: Add to queue if socket is not open
+		// Check if we have an active websocket connection
+		if (this.#transport) {
+			// If we have an active connection, use the websocket RPC
+			const rpcId = this.#rpcIdCounter;
+			this.#rpcIdCounter += 1;
 
-		const rpcId = this.#rpcIdCounter;
-		this.#rpcIdCounter += 1;
+			const { promise, resolve, reject } =
+				Promise.withResolvers<wsToClient.RpcResponseOk>();
+			this.#rpcInFlight.set(rpcId, { name, resolve, reject });
 
-		const { promise, resolve, reject } =
-			Promise.withResolvers<wsToClient.RpcResponseOk>();
-		this.#rpcInFlight.set(rpcId, { name, resolve, reject });
-
-		this.#sendMessage({
-			b: {
-				rr: {
-					i: rpcId,
-					n: name,
-					a: args,
+			this.#sendMessage({
+				b: {
+					rr: {
+						i: rpcId,
+						n: name,
+						a: args,
+					},
 				},
-			},
-		} satisfies wsToServer.ToServer);
+			} satisfies wsToServer.ToServer);
 
-		// TODO: Throw error if disconnect is called
+			// TODO: Throw error if disconnect is called
 
-		const { i: responseId, o: output } = await promise;
-		if (responseId !== rpcId)
-			throw new Error(
-				`Request ID ${rpcId} does not match response ID ${responseId}`,
-			);
+			const { i: responseId, o: output } = await promise;
+			if (responseId !== rpcId)
+				throw new Error(
+					`Request ID ${rpcId} does not match response ID ${responseId}`,
+				);
+
+			return output as Response;
+		} else {
+			// If no websocket connection, use HTTP RPC via manager
+			try {
+				// Get the manager endpoint from the endpoint provided
+				const managerEndpoint = this.endpoint.split('/manager/')[0];
+				const actorQueryStr = encodeURIComponent(JSON.stringify(this.actorQuery));
+				
+				const url = `${managerEndpoint}/actor/rpc/${name}?query=${actorQueryStr}`;
+				logger().debug("=== CLIENT HTTP RPC: Sending request ===", { 
+					url, 
+					managerEndpoint,
+					actorQuery: this.actorQuery, 
+					name,
+					args 
+				});
+				
+				try {
+					const response = await fetch(url, {
+						method: "POST",
+						headers: {
+							"Content-Type": "application/json",
+						},
+						body: JSON.stringify({
+							a: args,
+						}),
+					});
 
-		return output as Response;
+					logger().debug("=== CLIENT HTTP RPC: Response received ===", { 
+						status: response.status, 
+						ok: response.ok,
+						headers: Object.fromEntries([...response.headers])
+					});
+
+					if (!response.ok) {
+						try {
+							const errorData = await response.json();
+							logger().error("=== CLIENT HTTP RPC: Error response ===", { errorData });
+							throw new errors.ActionError(
+								errorData.c || "RPC_ERROR",
+								errorData.m || "RPC call failed",
+								errorData.md,
+							);
+						} catch (parseError) {
+							// If response is not JSON, get it as text and throw generic error
+							const errorText = await response.text();
+							logger().error("=== CLIENT HTTP RPC: Error parsing response ===", { 
+								errorText, 
+								parseError 
+							});
+							throw new errors.ActionError(
+								"RPC_ERROR",
+								`RPC call failed: ${errorText}`,
+								{},
+							);
+						}
+					}
+
+					// Clone response to avoid consuming it
+					const responseClone = response.clone();
+					const responseText = await responseClone.text();
+					logger().debug("=== CLIENT HTTP RPC: Response body ===", { responseText });
+
+					// Parse response body
+					try {
+						const responseData = JSON.parse(responseText);
+						logger().debug("=== CLIENT HTTP RPC: Parsed response ===", { responseData });
+						return responseData.o as Response;
+					} catch (parseError) {
+						logger().error("=== CLIENT HTTP RPC: Error parsing JSON ===", { 
+							responseText, 
+							parseError 
+						});
+						throw new errors.ActionError(
+							"RPC_ERROR",
+							`Failed to parse response: ${parseError}`,
+							{ responseText }
+						);
+					}
+				} catch (fetchError) {
+					logger().error("=== CLIENT HTTP RPC: Fetch error ===", { 
+						error: fetchError,
+						url
+					});
+					throw new errors.ActionError(
+						"RPC_ERROR",
+						`Fetch failed: ${fetchError}`,
+						{ cause: fetchError }
+					);
+				}
+			} catch (error) {
+				if (error instanceof errors.ActionError) {
+					throw error;
+				}
+				throw new errors.ActionError(
+					"RPC_ERROR", 
+					`Failed to execute RPC ${name}: ${error}`, 
+					{ cause: error }
+				);
+			}
+		}
 	}
 
 	//async #rpcHttp<Args extends Array<unknown> = unknown[], Response = unknown>(name: string, ...args: Args): Promise<Response> {
@@ -453,7 +555,17 @@ enc
 	}
 
 	#buildConnUrl(transport: Transport): string {
-		let url = `${this.endpoint}/connect/${transport}?encoding=${this.encodingKind}`;
+		// Get the manager endpoint from the endpoint provided
+		const managerEndpoint = this.endpoint.split('/manager/')[0];
+		const actorQueryStr = encodeURIComponent(JSON.stringify(this.actorQuery));
+		
+		logger().debug("=== Client building conn URL ===", {
+			originalEndpoint: this.endpoint,
+			managerEndpoint: managerEndpoint,
+			transport: transport
+		});
+		
+		let url = `${managerEndpoint}/actor/connect/${transport}?encoding=${this.encodingKind}&query=${actorQueryStr}`;
 
 		if (this.params !== undefined) {
 			const paramsStr = JSON.stringify(this.params);
@@ -469,6 +581,8 @@ enc
 		if (transport === "websocket") {
 			url = url.replace(/^http:/, "ws:").replace(/^https:/, "wss:");
 		}
+		
+		logger().debug("=== Client final conn URL ===", { url });
 
 		return url;
 	}
@@ -617,7 +731,11 @@ enc
 			if (!this.#connectionId || !this.#connectionToken)
 				throw new errors.InternalError("Missing connection ID or token.");
 
-			let url = `${this.endpoint}/connections/${this.#connectionId}/message?encoding=${this.encodingKind}&connectionToken=${encodeURIComponent(this.#connectionToken)}`;
+			// Get the manager endpoint from the endpoint provided
+			const managerEndpoint = this.endpoint.split('/manager/')[0];
+			const actorQueryStr = encodeURIComponent(JSON.stringify(this.actorQuery));
+
+			let url = `${managerEndpoint}/actor/connections/${this.#connectionId}/message?encoding=${this.encodingKind}&connectionToken=${encodeURIComponent(this.#connectionToken)}&query=${actorQueryStr}`;
 
 			// TODO: Implement ordered messages, this is not guaranteed order. Needs to use an index in order to ensure we can pipeline requests efficiently.
 			// TODO: Validate that we're using HTTP/3 whenever possible for pipelining requests

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

@@ -212,21 +212,18 @@ export class ClientRaw {
 			params: opts?.params,
 		});
 
-		const resJson = await this.#sendManagerRequest<
-			ActorsRequest,
-			ActorsResponse
-		>("POST", "/manager/actors", {
-			query: {
-				getForId: {
-					actorId,
-				},
+		const actorQuery = {
+			getForId: {
+				actorId,
 			},
-		});
+		};
 
+		const managerEndpoint = await this.#managerEndpointPromise;
 		const conn = await this.#createConn(
-			resJson.endpoint,
+			managerEndpoint,
 			opts?.params,
-			resJson.supportedTransports,
+			["websocket", "sse"],
+			actorQuery
 		);
 		return this.#createProxy(conn) as ActorConn<AD>;
 	}
@@ -284,18 +281,18 @@ export class ClientRaw {
 			create,
 		});
 
-		let requestQuery;
+		let actorQuery;
 		if (opts?.noCreate) {
 			// Use getForKey endpoint if noCreate is specified
-			requestQuery = {
+			actorQuery = {
 				getForKey: {
 					name,
 					key: keyArray,
 				},
 			};
 		} else {
 			// Use getOrCreateForKey endpoint
-			requestQuery = {
+			actorQuery = {
 				getOrCreateForKey: {
 					name,
 					key: keyArray,
@@ -304,17 +301,12 @@ export class ClientRaw {
 			};
 		}
 
-		const resJson = await this.#sendManagerRequest<
-			ActorsRequest,
-			ActorsResponse
-		>("POST", "/manager/actors", {
-			query: requestQuery,
-		});
-
+		const managerEndpoint = await this.#managerEndpointPromise;
 		const conn = await this.#createConn(
-			resJson.endpoint,
+			managerEndpoint,
 			opts?.params,
-			resJson.supportedTransports,
+			["websocket", "sse"],
+			actorQuery
 		);
 		return this.#createProxy(conn) as ActorConn<AD>;
 	}
@@ -374,19 +366,16 @@ export class ClientRaw {
 			create,
 		});
 
-		const resJson = await this.#sendManagerRequest<
-			ActorsRequest,
-			ActorsResponse
-		>("POST", "/manager/actors", {
-			query: {
-				create,
-			},
-		});
+		const actorQuery = {
+			create,
+		};
 
+		const managerEndpoint = await this.#managerEndpointPromise;
 		const conn = await this.#createConn(
-			resJson.endpoint,
+			managerEndpoint,
 			opts?.params,
-			resJson.supportedTransports,
+			["websocket", "sse"],
+			actorQuery
 		);
 		return this.#createProxy(conn) as ActorConn<AD>;
 	}
@@ -395,6 +384,7 @@ export class ClientRaw {
 		endpoint: string,
 		params: unknown,
 		serverTransports: Transport[],
+		actorQuery: unknown,
 	): Promise<ActorConnRaw> {
 		const imports = await this.#dynamicImportsPromise;
 
@@ -406,6 +396,7 @@ export class ClientRaw {
 			this.#supportedTransports,
 			serverTransports,
 			imports,
+			actorQuery,
 		);
 		this[ACTOR_CONNS_SYMBOL].add(conn);
 		conn[CONNECT_SYMBOL]();