|
| 1 | +import * as errors from "@/actor/errors"; |
| 2 | +import * as protoHttpAction from "@/actor/protocol/http/action"; |
| 3 | +import { logger } from "./log"; |
| 4 | +import type { EventSource } from "eventsource"; |
| 5 | +import type * as wsToServer from "@/actor/protocol/message/to-server"; |
| 6 | +import { type Encoding, serialize } from "@/actor/protocol/serde"; |
| 7 | +import { |
| 8 | + HEADER_CONN_PARAMS, |
| 9 | + HEADER_ENCODING, |
| 10 | + type ConnectionHandlers, |
| 11 | +} from "@/actor/router-endpoints"; |
| 12 | +import { HonoRequest, type Context as HonoContext, type Next } from "hono"; |
| 13 | +import invariant from "invariant"; |
| 14 | +import { ClientDriver } from "@/client/client"; |
| 15 | +import { ManagerDriver } from "@/manager/driver"; |
| 16 | +import { ActorQuery } from "@/manager/protocol/query"; |
| 17 | +import { ConnRoutingHandler } from "@/actor/conn-routing-handler"; |
| 18 | +import { sendHttpRequest, serializeWithEncoding } from "@/client/utils"; |
| 19 | +import { ActionRequest, ActionResponse } from "@/actor/protocol/http/action"; |
| 20 | +import { assertUnreachable } from "@/actor/utils"; |
| 21 | + |
| 22 | +/** |
| 23 | + * Client driver that calls the manager driver inline. |
| 24 | + * |
| 25 | + * This driver can access private resources. |
| 26 | + * |
| 27 | + * This driver serves a double purpose as: |
| 28 | + * - Providing the client for the internal requests |
| 29 | + * - Provide the driver for the manager HTTP router (see manager/router.ts) |
| 30 | + */ |
| 31 | +export function createInlineClientDriver( |
| 32 | + managerDriver: ManagerDriver, |
| 33 | + routingHandler: ConnRoutingHandler, |
| 34 | +): ClientDriver { |
| 35 | + //// Lazily import the dynamic imports so we don't have to turn `createClient` in to an aysnc fn |
| 36 | + //const dynamicImports = (async () => { |
| 37 | + // // Import dynamic dependencies |
| 38 | + // const [WebSocket, EventSource] = await Promise.all([ |
| 39 | + // importWebSocket(), |
| 40 | + // importEventSource(), |
| 41 | + // ]); |
| 42 | + // return { |
| 43 | + // WebSocket, |
| 44 | + // EventSource, |
| 45 | + // }; |
| 46 | + //})(); |
| 47 | + |
| 48 | + const driver: ClientDriver = { |
| 49 | + action: async <Args extends Array<unknown> = unknown[], Response = unknown>( |
| 50 | + req: HonoRequest | undefined, |
| 51 | + actorQuery: ActorQuery, |
| 52 | + encoding: Encoding, |
| 53 | + params: unknown, |
| 54 | + actionName: string, |
| 55 | + ...args: Args |
| 56 | + ): Promise<Response> => { |
| 57 | + // Get the actor ID and meta |
| 58 | + const { actorId, meta } = await queryActor( |
| 59 | + req, |
| 60 | + actorQuery, |
| 61 | + managerDriver, |
| 62 | + ); |
| 63 | + logger().debug("found actor for action", { actorId, meta }); |
| 64 | + invariant(actorId, "Missing actor ID"); |
| 65 | + |
| 66 | + // Invoke the action |
| 67 | + logger().debug("handling action", { actionName, encoding }); |
| 68 | + if ("inline" in routingHandler) { |
| 69 | + const { output } = await routingHandler.inline.handlers.onAction({ |
| 70 | + req, |
| 71 | + params, |
| 72 | + actionName, |
| 73 | + actionArgs: args, |
| 74 | + actorId, |
| 75 | + }); |
| 76 | + return output as Response; |
| 77 | + } else if ("custom" in routingHandler) { |
| 78 | + const responseData = await sendHttpRequest< |
| 79 | + ActionRequest, |
| 80 | + ActionResponse |
| 81 | + >({ |
| 82 | + url: `http://actor/action/${encodeURIComponent(actionName)}`, |
| 83 | + method: "POST", |
| 84 | + headers: { |
| 85 | + [HEADER_ENCODING]: encoding, |
| 86 | + ...(params !== undefined |
| 87 | + ? { [HEADER_CONN_PARAMS]: JSON.stringify(params) } |
| 88 | + : {}), |
| 89 | + }, |
| 90 | + body: { a: args } satisfies ActionRequest, |
| 91 | + encoding: encoding, |
| 92 | + customFetch: routingHandler.custom.sendRequest.bind( |
| 93 | + undefined, |
| 94 | + actorId, |
| 95 | + meta, |
| 96 | + ), |
| 97 | + }); |
| 98 | + |
| 99 | + return responseData.o as Response; |
| 100 | + } else { |
| 101 | + assertUnreachable(routingHandler); |
| 102 | + } |
| 103 | + }, |
| 104 | + |
| 105 | + resolveActorId: async ( |
| 106 | + req: HonoRequest | undefined, |
| 107 | + actorQuery: ActorQuery, |
| 108 | + _encodingKind: Encoding, |
| 109 | + ): Promise<string> => { |
| 110 | + // Get the actor ID and meta |
| 111 | + const { actorId } = await queryActor(req, actorQuery, managerDriver); |
| 112 | + logger().debug("resolved actor", { actorId }); |
| 113 | + invariant(actorId, "missing actor ID"); |
| 114 | + |
| 115 | + return actorId; |
| 116 | + }, |
| 117 | + |
| 118 | + connectWebSocket: async ( |
| 119 | + req: HonoRequest | undefined, |
| 120 | + actorQuery: ActorQuery, |
| 121 | + encodingKind: Encoding, |
| 122 | + ): Promise<WebSocket> => { |
| 123 | + throw "UNIMPLEMENTED"; |
| 124 | + }, |
| 125 | + |
| 126 | + connectSse: async ( |
| 127 | + req: HonoRequest | undefined, |
| 128 | + actorQuery: ActorQuery, |
| 129 | + encodingKind: Encoding, |
| 130 | + params: unknown, |
| 131 | + ): Promise<EventSource> => { |
| 132 | + throw "UNIMPLEMENTED"; |
| 133 | + }, |
| 134 | + |
| 135 | + sendHttpMessage: async ( |
| 136 | + req: HonoRequest | undefined, |
| 137 | + actorId: string, |
| 138 | + encoding: Encoding, |
| 139 | + connectionId: string, |
| 140 | + connectionToken: string, |
| 141 | + message: wsToServer.ToServer, |
| 142 | + ): Promise<Response> => { |
| 143 | + throw "UNIMPLEMENTED"; |
| 144 | + }, |
| 145 | + }; |
| 146 | + |
| 147 | + return driver; |
| 148 | +} |
| 149 | + |
| 150 | +/** |
| 151 | + * Query the manager driver to get or create an actor based on the provided query |
| 152 | + */ |
| 153 | +export async function queryActor( |
| 154 | + req: HonoRequest | undefined, |
| 155 | + query: ActorQuery, |
| 156 | + driver: ManagerDriver, |
| 157 | +): Promise<{ actorId: string; meta?: unknown }> { |
| 158 | + logger().debug("querying actor", { query }); |
| 159 | + let actorOutput: { actorId: string; meta?: unknown }; |
| 160 | + if ("getForId" in query) { |
| 161 | + const output = await driver.getForId({ |
| 162 | + req, |
| 163 | + actorId: query.getForId.actorId, |
| 164 | + }); |
| 165 | + if (!output) throw new errors.ActorNotFound(query.getForId.actorId); |
| 166 | + actorOutput = output; |
| 167 | + } else if ("getForKey" in query) { |
| 168 | + const existingActor = await driver.getWithKey({ |
| 169 | + req, |
| 170 | + name: query.getForKey.name, |
| 171 | + key: query.getForKey.key, |
| 172 | + }); |
| 173 | + if (!existingActor) { |
| 174 | + throw new errors.ActorNotFound( |
| 175 | + `${query.getForKey.name}:${JSON.stringify(query.getForKey.key)}`, |
| 176 | + ); |
| 177 | + } |
| 178 | + actorOutput = existingActor; |
| 179 | + } else if ("getOrCreateForKey" in query) { |
| 180 | + const getOrCreateOutput = await driver.getOrCreateWithKey({ |
| 181 | + req, |
| 182 | + name: query.getOrCreateForKey.name, |
| 183 | + key: query.getOrCreateForKey.key, |
| 184 | + input: query.getOrCreateForKey.input, |
| 185 | + region: query.getOrCreateForKey.region, |
| 186 | + }); |
| 187 | + actorOutput = { |
| 188 | + actorId: getOrCreateOutput.actorId, |
| 189 | + meta: getOrCreateOutput.meta, |
| 190 | + }; |
| 191 | + } else if ("create" in query) { |
| 192 | + const createOutput = await driver.createActor({ |
| 193 | + req, |
| 194 | + name: query.create.name, |
| 195 | + key: query.create.key, |
| 196 | + input: query.create.input, |
| 197 | + region: query.create.region, |
| 198 | + }); |
| 199 | + actorOutput = { |
| 200 | + actorId: createOutput.actorId, |
| 201 | + meta: createOutput.meta, |
| 202 | + }; |
| 203 | + } else { |
| 204 | + throw new errors.InvalidRequest("Invalid query format"); |
| 205 | + } |
| 206 | + |
| 207 | + logger().debug("actor query result", { |
| 208 | + actorId: actorOutput.actorId, |
| 209 | + meta: actorOutput.meta, |
| 210 | + }); |
| 211 | + return { actorId: actorOutput.actorId, meta: actorOutput.meta }; |
| 212 | +} |
0 commit comments