| 1 | import type { ConfigError } from "effect/Config"; |
| 2 | import * as Context from "effect/Context"; |
| 3 | import * as Effect from "effect/Effect"; |
| 4 | import * as Layer from "effect/Layer"; |
| 5 | import * as HttpClient from "effect/http/HttpClient"; |
| 6 | import * as HttpClientResponse from "effect/http/HttpClientResponse"; |
| 7 | import { |
| 8 | type Rpc, |
| 9 | RpcClient, |
| 10 | type RpcGroup, |
| 11 | RpcSerialization, |
| 12 | } from "effect/rpc"; |
| 13 | import type * as RpcClientError from "effect/rpc/RpcClientError"; |
| 14 | import type { Dependencies } from "../../Dependencies.ts"; |
| 15 | import type { HttpEffect } from "../../Http.ts"; |
| 16 | import type { InputProps } from "../../Input.ts"; |
| 17 | import type { Named } from "../../Named.ts"; |
| 18 | import type { PlatformIdentity } from "../../Platform.ts"; |
| 19 | import type { Rpc as RpcShape } from "../../Rpc.ts"; |
| 20 | import { effectClass, taggedFunction } from "../../Util/effect.ts"; |
| 21 | import type { Worker, WorkerProps } from "./Worker.ts"; |
| 22 | import { Worker as WorkerCtor, WorkerEnvironment } from "./Worker.ts"; |
| 23 | |
| 24 | /** |
| 25 | * Props for {@link RpcWorker}. Same shape as {@link WorkerProps} with |
| 26 | * an additional `schema` field carrying the rpc group definition. |
| 27 | */ |
| 28 | export type RpcWorkerProps<Rpcs extends Rpc.Any> = { |
| 29 | /** |
| 30 | * The {@link RpcGroup.RpcGroup} served on this worker's `fetch` |
| 31 | * handler. The same value should be importable by any consumer of |
| 32 | * the worker (other workers or scripts using `RpcClient.make`). |
| 33 | */ |
| 34 | readonly schema: RpcGroup.RpcGroup<Rpcs>; |
| 35 | }; |
| 36 | |
| 37 | // Context tag carrying the surrounding `RpcWorker` value, so |
| 38 | // `yield* RpcWorker` inside the impl resolves to the worker itself |
| 39 | // (mirrors `yield* Worker` semantics inside `Cloudflare.Worker`). |
| 40 | // Documented as part of the main `RpcWorker` JSDoc below. |
| 41 | export class RpcWorkerScope extends Context.Service<RpcWorkerScope, Worker>()( |
| 42 | "Cloudflare.RpcWorker", |
| 43 | ) {} |
| 44 | |
| 45 | // Symbol used internally to stash the rpc `RpcGroup` schema on the |
| 46 | // `effectClass` that `RpcWorker(...)` returns, so `RpcWorker.bind` |
| 47 | // can recover it when handed the class later. |
| 48 | const SchemaSymbol = Symbol.for("alchemy.RpcWorker.schema"); |
| 49 | |
| 50 | // Phantom carrier — `RpcWorker.bind` uses `Rpcs` to type the returned |
| 51 | // rpc client. We don't tie it to a specific `Effect.R` since the |
| 52 | // requirements come from the underlying `Cloudflare.Worker` factory |
| 53 | // and shouldn't leak into the Stack-side service signature. |
| 54 | // |
| 55 | // `Deps` mirrors `Cloudflare.Worker<Self, Bindings, Deps>` — declares |
| 56 | // the DOs / Workers this script publishes for cross-script binding so |
| 57 | // `Counter.from(WorkerA)` type-checks from another script. |
| 58 | // |
| 59 | // `Id` is the const logical id (`Named<Id>`), matching `Cloudflare.Worker`. |
| 60 | export interface RpcWorkerYieldable< |
| 61 | Self, |
| 62 | Rpcs extends Rpc.Any, |
| 63 | Deps = never, |
| 64 | Id extends string = string, |
| 65 | > |
| 66 | extends |
| 67 | Effect.Effect<Worker<{}> & RpcShape<Self> & Dependencies<Deps>, never, any>, |
| 68 | Named<Id>, |
| 69 | PlatformIdentity<Id> { |
| 70 | /** @internal */ |
| 71 | readonly [SchemaSymbol]: RpcGroup.RpcGroup<Rpcs>; |
| 72 | } |
| 73 | |
| 74 | /** |
| 75 | * Type of the {@link RpcWorker} constructor. Mirrors the class-form |
| 76 | * signature of {@link Worker} so `class X extends RpcWorker<X>()(...)` |
| 77 | * works identically and the resulting worker is `Rpc<Self>`-typed for |
| 78 | * binding consumers. |
| 79 | */ |
| 80 | export interface RpcWorkerClass extends Effect.Effect< |
| 81 | Worker, |
| 82 | never, |
| 83 | RpcWorkerScope |
| 84 | > { |
| 85 | /** |
| 86 | * Class-based form: `class X extends RpcWorker<X>()(name, props, impl)`. |
| 87 | * |
| 88 | * The optional second type argument `Deps` mirrors |
| 89 | * `Cloudflare.Worker<Self, Bindings, Deps>` — it declares the DOs |
| 90 | * this Worker publishes for cross-script binding so consumers can |
| 91 | * write `Counter.from(WorkerA)` and have it type-check. `Rpcs` is |
| 92 | * always inferred from `props.schema`. |
| 93 | * |
| 94 | * Yielding the class in a Stack returns the {@link Worker} resource |
| 95 | * (so `worker.url`, `worker.workerName`, etc. work as usual). To get |
| 96 | * a typed `RpcClient` for *this* worker's rpc group from inside |
| 97 | * another worker's init, call {@link RpcWorker.bind}. |
| 98 | */ |
| 99 | <Self, Deps = never>(): { |
| 100 | /** |
| 101 | * Modular form (no impl). Mirrors `Cloudflare.Worker<Self>()(id, props)`. |
| 102 | * Use `WorkerClass.make(impl)` to provide the runtime as a |
| 103 | * `Layer.Layer<Self>` so consumers that don't host the worker can |
| 104 | * import the class without pulling its runtime into their bundle. |
| 105 | */ |
| 106 | <const Id extends string, Rpcs extends Rpc.Any>( |
| 107 | id: Id, |
| 108 | props: RpcWorkerProps<Rpcs>, |
| 109 | ): RpcWorkerYieldable<Self, Rpcs, Deps, Id> & { |
| 110 | new (_: never): Named<Id>; |
| 111 | make<InnerR = never, InitReq = never>( |
| 112 | props: InputProps<WorkerProps>, |
| 113 | impl: Effect.Effect< |
| 114 | Effect.Effect<HttpEffect<InnerR>, never, InnerR>, |
| 115 | ConfigError, |
| 116 | InitReq |
| 117 | >, |
| 118 | ): Layer.Layer<Self, never, Exclude<InitReq | InnerR, never>>; |
| 119 | }; |
| 120 | /** Inline-impl form. */ |
| 121 | < |
| 122 | const Id extends string, |
| 123 | Rpcs extends Rpc.Any, |
| 124 | InnerR = never, |
| 125 | InitReq = never, |
| 126 | >( |
| 127 | id: Id, |
| 128 | props: RpcWorkerProps<Rpcs> & InputProps<WorkerProps>, |
| 129 | impl: Effect.Effect< |
| 130 | Effect.Effect<HttpEffect<InnerR>, never, InnerR>, |
| 131 | ConfigError, |
| 132 | InitReq |
| 133 | >, |
| 134 | ): RpcWorkerYieldable<Self, Rpcs, Deps, Id> & { |
| 135 | // Phantom — `class X extends RpcWorker<X>()(...)` carries `X` |
| 136 | // through the result type via `Rpc<Self>` on the binding side; |
| 137 | // the instance shape itself is empty besides `Named<Id>` (no |
| 138 | // methods exposed on `new X(...)` — everything goes through |
| 139 | // `fetch`). |
| 140 | new (_: never): Named<Id>; |
| 141 | }; |
| 142 | }; |
| 143 | |
| 144 | /** |
| 145 | * Bind a typed Effect rpc client to a worker resource, using the |
| 146 | * worker's declared rpc {@link RpcGroup.RpcGroup} schema. Mirrors |
| 147 | * `Cloudflare.R2.ReadWriteBucket(MyBucket)` and friends. |
| 148 | * |
| 149 | * Yield once at **init** — the result is a normal `RpcClient` you |
| 150 | * can call directly from any per-request handler. Internally each |
| 151 | * method invocation builds a fresh underlying `RpcClient` (through |
| 152 | * a Proxy) because Cloudflare rejects I/O objects created on a |
| 153 | * previous request; this is hidden from the consumer. |
| 154 | * |
| 155 | * Pair with {@link RpcWorker} on the server side; both ends share |
| 156 | * the same schema so values round-trip through one `Schema` codec. |
| 157 | * |
| 158 | * @example |
| 159 | * ```ts |
| 160 | * // INIT: register the binding once and get the typed client |
| 161 | * const tasks = yield* Cloudflare.RpcWorker.bind(TaskWorker); |
| 162 | * |
| 163 | * // PER-REQUEST: call methods directly |
| 164 | * proxyGetTask: ({ id }) => tasks.getTask({ id }), |
| 165 | * ``` |
| 166 | */ |
| 167 | readonly bind: <Self, Rpcs extends Rpc.Any>( |
| 168 | workerEff: RpcWorkerYieldable<Self, Rpcs>, |
| 169 | ) => Effect.Effect< |
| 170 | RpcClient.RpcClient<Rpcs, RpcClientError.RpcClientError> & RpcShape<Self>, |
| 171 | never, |
| 172 | Worker |
| 173 | >; |
| 174 | } |
| 175 | |
| 176 | /** @internal helper — exposed as `RpcWorker.bind` below. */ |
| 177 | const bind = <Self, Rpcs extends Rpc.Any>( |
| 178 | workerEff: RpcWorkerYieldable<Self, Rpcs>, |
| 179 | ): Effect.Effect< |
| 180 | RpcClient.RpcClient<Rpcs, RpcClientError.RpcClientError> & RpcShape<Self>, |
| 181 | never, |
| 182 | Worker |
| 183 | > => |
| 184 | Effect.gen(function* () { |
| 185 | const schema = (workerEff as unknown as Record<symbol, unknown>)[ |
| 186 | SchemaSymbol |
| 187 | ] as RpcGroup.RpcGroup<Rpcs> | undefined; |
| 188 | if (!schema) { |
| 189 | return yield* Effect.die( |
| 190 | new Error( |
| 191 | "RpcWorker.bind: passed value isn't an RpcWorker class — no schema attached.", |
| 192 | ), |
| 193 | ); |
| 194 | } |
| 195 | const worker = (yield* workerEff) as Worker; |
| 196 | // Register the service binding on the surrounding worker at INIT. |
| 197 | // Mirrors `Cloudflare.Workers.bindWorker` — yielding the class is *not* |
| 198 | // enough; we need an explicit `self.bind\`${worker}\`(...)` so |
| 199 | // workerd surfaces the stub on `env` at request time. |
| 200 | const self = yield* WorkerCtor; |
| 201 | yield* self.bind`${worker}`({ |
| 202 | bindings: [ |
| 203 | { |
| 204 | type: "service", |
| 205 | name: worker.LogicalId, |
| 206 | service: worker.workerName, |
| 207 | }, |
| 208 | ], |
| 209 | }); |
| 210 | |
| 211 | // Build a per-call factory that wraps the Cloudflare service- |
| 212 | // binding stub (Promise-based fetch) into an Effect `HttpClient`, |
| 213 | // then `RpcClient.make` over that transport. Each call gets its |
| 214 | // own `RpcClient` because Cloudflare rejects I/O objects (the |
| 215 | // stub.fetch body) that were created on a previous request. |
| 216 | const makeFreshClient = Effect.gen(function* () { |
| 217 | const env = yield* WorkerEnvironment; |
| 218 | const stub = (env as Record<string, { fetch: typeof fetch }>)[ |
| 219 | worker.LogicalId |
| 220 | ]; |
| 221 | const httpClient = HttpClient.make((req) => |
| 222 | Effect.promise((signal) => |
| 223 | stub.fetch( |
| 224 | new Request(req.url, { |
| 225 | method: req.method, |
| 226 | headers: new Headers(req.headers as any), |
| 227 | body: (req.body as any)?.body ?? undefined, |
| 228 | signal, |
| 229 | }), |
| 230 | ), |
| 231 | ).pipe(Effect.map((res) => HttpClientResponse.fromWeb(req, res))), |
| 232 | ); |
| 233 | const protocol = RpcClient.layerProtocolHttp({ |
| 234 | url: "http://alchemy-rpc-worker/", |
| 235 | }).pipe( |
| 236 | Layer.provide(RpcSerialization.layerNdjson), |
| 237 | Layer.provide(Layer.succeed(HttpClient.HttpClient, httpClient)), |
| 238 | ); |
| 239 | return yield* RpcClient.make(schema).pipe(Effect.provide(protocol)); |
| 240 | }); |
| 241 | |
| 242 | // Return a Proxy that LOOKS like an `RpcClient<Rpcs>` but defers |
| 243 | // the actual `RpcClient` construction to call time. Mirrors |
| 244 | // `makeRpcStub` from `./Rpc.ts`: each method invocation builds a |
| 245 | // fresh client, runs the call inside `Effect.scoped`, and tears |
| 246 | // the client's scope down with the call. From the consumer's |
| 247 | // perspective this is a normal `RpcClient`: |
| 248 | // |
| 249 | // const tasks = yield* Cloudflare.RpcWorker.bind(TaskWorker); |
| 250 | // yield* tasks.getTask({ id: "abc" }); |
| 251 | return new Proxy( |
| 252 | {}, |
| 253 | { |
| 254 | get: (_target, method) => { |
| 255 | if (typeof method !== "string") return undefined; |
| 256 | return (...args: unknown[]) => |
| 257 | Effect.gen(function* () { |
| 258 | const client = yield* makeFreshClient; |
| 259 | const fn = ( |
| 260 | client as unknown as Record< |
| 261 | string, |
| 262 | (...a: unknown[]) => Effect.Effect<unknown, unknown, unknown> |
| 263 | > |
| 264 | )[method]; |
| 265 | return yield* fn(...args); |
| 266 | }).pipe(Effect.scoped); |
| 267 | }, |
| 268 | }, |
| 269 | ) as RpcClient.RpcClient<Rpcs, RpcClientError.RpcClientError> & |
| 270 | RpcShape<Self>; |
| 271 | }); |
| 272 | |
| 273 | /** |
| 274 | * `RpcWorker` is a thin sugar over {@link Worker} for the common case |
| 275 | * where a worker's entire `fetch` surface is a typed Effect `RpcGroup`. |
| 276 | * It takes the rpc `schema` directly in props alongside `main`, and |
| 277 | * accepts an init Effect that returns the already-piped |
| 278 | * `RpcServer.toHttpEffect(...)`-producing Effect (no `{ fetch }` |
| 279 | * wrapper) — the wrapper plugs it into the worker's `fetch` for you. |
| 280 | * |
| 281 | * Functionally identical to writing `Cloudflare.Worker(...)` with |
| 282 | * `return { fetch: RpcServer.toHttpEffect(schema).pipe(...) }`; use |
| 283 | * whichever style you prefer. |
| 284 | * |
| 285 | * The class form (`class X extends Cloudflare.RpcWorker<X>()(...)`) |
| 286 | * carries `Self` through the result type as `Rpc<Self>`, so other |
| 287 | * workers binding to this one see the rpc shape pinned to `Self`. |
| 288 | * |
| 289 | * |
| 290 | * ### Defining the rpc group |
| 291 | * **Example:** Pure schema description |
| 292 | * The rpc group and its schemas live outside any worker so both the |
| 293 | * server (`RpcWorker`) and any consumers (`RpcClient.make` / |
| 294 | * `RpcDurableObject`) import the same value. |
| 295 | * ```typescript |
| 296 | * import * as Schema from "effect/Schema"; |
| 297 | * import { Rpc, RpcGroup } from "effect/rpc"; |
| 298 | * |
| 299 | * export class TaskNotFound extends Schema.TaggedClass<TaskNotFound>()( |
| 300 | * "TaskNotFound", |
| 301 | * { id: Schema.String }, |
| 302 | * ) {} |
| 303 | * |
| 304 | * const getTask = Rpc.make("getTask", { |
| 305 | * payload: { id: Schema.String }, |
| 306 | * success: Schema.String, |
| 307 | * error: TaskNotFound, |
| 308 | * }); |
| 309 | * |
| 310 | * export class TaskRpcs extends RpcGroup.make(getTask) {} |
| 311 | * ``` |
| 312 | * |
| 313 | * ### Implementing the worker |
| 314 | * **Example:** Class form (recommended) |
| 315 | * Mirrors `Cloudflare.Worker<Self>()(...)` — `class X extends ...` |
| 316 | * works the same. The init Effect builds a handlers `Layer` from the |
| 317 | * group and returns the `RpcServer.toHttpEffect(schema)`-piped Effect |
| 318 | * directly. |
| 319 | * ```typescript |
| 320 | * import * as Cloudflare from "alchemy/Cloudflare"; |
| 321 | * import * as Effect from "effect/Effect"; |
| 322 | * import * as Layer from "effect/Layer"; |
| 323 | * import { RpcSerialization, RpcServer } from "effect/rpc"; |
| 324 | * import { TaskRpcs } from "./rpcs.ts"; |
| 325 | * |
| 326 | * export default class Worker extends Cloudflare.RpcWorker<Worker>()( |
| 327 | * "Worker", |
| 328 | * { main: import.meta.url, schema: TaskRpcs }, |
| 329 | * Effect.gen(function* () { |
| 330 | * const handlers = TaskRpcs.toLayer({ |
| 331 | * getTask: ({ id }) => Effect.succeed(`task-${id}`), |
| 332 | * }); |
| 333 | * return RpcServer.toHttpEffect(TaskRpcs).pipe( |
| 334 | * Effect.provide(Layer.mergeAll(handlers, RpcSerialization.layerJson)), |
| 335 | * ); |
| 336 | * }), |
| 337 | * ) {} |
| 338 | * ``` |
| 339 | * |
| 340 | * **Example:** NDJSON for streaming rpcs |
| 341 | * If any rpc in the group is a streaming rpc, the wire serialization |
| 342 | * must be `RpcSerialization.layerNdjson` — streaming rpcs need |
| 343 | * newline framing on the wire. |
| 344 | * ```typescript |
| 345 | * return RpcServer.toHttpEffect(ChatRpcs).pipe( |
| 346 | * Effect.provide(handlers), |
| 347 | * Effect.provide(RpcSerialization.layerNdjson), |
| 348 | * ); |
| 349 | * ``` |
| 350 | * |
| 351 | * ### Modular form: separate the class from its runtime |
| 352 | * **Example:** Class declaration with no impl + `static make(impl)` |
| 353 | * The inline class form above bundles the runtime into the class |
| 354 | * declaration. The two-arg form `(id, props)` declares the class |
| 355 | * as a pure tagged identifier; provide the runtime separately via |
| 356 | * `WorkerClass.make(impl)` so consumers can import the class for |
| 357 | * binding without pulling the host's runtime into their bundle. |
| 358 | * ```typescript |
| 359 | * export class TaskWorker extends Cloudflare.RpcWorker<TaskWorker>()( |
| 360 | * "TaskWorker", |
| 361 | * { main: import.meta.url, schema: TaskRpcs }, |
| 362 | * ) {} |
| 363 | * |
| 364 | * // Only the host script imports this default export; consumers |
| 365 | * // import the class above for `RpcWorker.bind(TaskWorker)`. |
| 366 | * export default TaskWorker.make( |
| 367 | * Effect.gen(function* () { |
| 368 | * const handlers = TaskRpcs.toLayer({ |
| 369 | * getTask: ({ id }) => Effect.succeed(`task-${id}`), |
| 370 | * }); |
| 371 | * return RpcServer.toHttpEffect(TaskRpcs).pipe( |
| 372 | * Effect.provide(Layer.mergeAll(handlers, RpcSerialization.layerJson)), |
| 373 | * ); |
| 374 | * }), |
| 375 | * ); |
| 376 | * ``` |
| 377 | * |
| 378 | * ### Hosting a Durable Object for cross-script binding |
| 379 | * **Example:** `RpcWorker<Self, Deps>()` declares published DOs |
| 380 | * The optional second type argument `Deps` mirrors |
| 381 | * `Cloudflare.Worker<Self, Bindings, Deps>` — it declares the DOs |
| 382 | * this Worker publishes for cross-script binding. With `Counter` |
| 383 | * named in `Deps`, any other Worker can write |
| 384 | * `Counter.from(TaskWorker)` and have it type-check. |
| 385 | * ```typescript |
| 386 | * import { Counter } from "./counter.ts"; |
| 387 | * |
| 388 | * export class TaskWorker extends Cloudflare.RpcWorker<TaskWorker, Counter>()( |
| 389 | * "TaskWorker", |
| 390 | * { main: import.meta.url, schema: TaskRpcs }, |
| 391 | * ) {} |
| 392 | * ``` |
| 393 | * See {@link RpcDurableObject} for the consumer side |
| 394 | * (`Counter.from(TaskWorker)`). |
| 395 | * |
| 396 | * ### Binding it from another worker |
| 397 | * **Example:** `Cloudflare.RpcWorker.bind(WorkerClass)` |
| 398 | * Inside another worker's init, `RpcWorker.bind(WorkerClass)` |
| 399 | * registers the service binding on the surrounding worker and returns |
| 400 | * a typed `RpcClient` you can call directly from any per-request |
| 401 | * handler. Internally each method invocation builds a fresh underlying |
| 402 | * client (because Cloudflare rejects cross-request reuse of the |
| 403 | * stub I/O), but that's hidden behind a Proxy so the consumer sees a |
| 404 | * normal `RpcClient`. |
| 405 | * ```typescript |
| 406 | * import TaskWorker from "./task-worker.ts"; |
| 407 | * |
| 408 | * export default class Caller extends Cloudflare.RpcWorker<Caller>()( |
| 409 | * "Caller", |
| 410 | * { main: import.meta.url, schema: CallerRpcs }, |
| 411 | * Effect.gen(function* () { |
| 412 | * // INIT: register binding, get the typed client |
| 413 | * const tasks = yield* Cloudflare.RpcWorker.bind(TaskWorker); |
| 414 | * |
| 415 | * const handlers = CallerRpcs.toLayer({ |
| 416 | * // PER-REQUEST: just call methods directly |
| 417 | * proxyGetTask: ({ id }) => tasks.getTask({ id }), |
| 418 | * }); |
| 419 | * return RpcServer.toHttpEffect(CallerRpcs).pipe( |
| 420 | * Effect.provide(Layer.mergeAll(handlers, RpcSerialization.layerJson)), |
| 421 | * ); |
| 422 | * }), |
| 423 | * ) {} |
| 424 | * ``` |
| 425 | * |
| 426 | * ### Driving it from a test |
| 427 | * **Example:** `Test.make` + `RpcClient.make` |
| 428 | * The same `RpcGroup` drives a typed client. `Test.make` deploys the |
| 429 | * stack once for the file; each test yields the deploy handle for its |
| 430 | * URL and calls procedures directly. |
| 431 | * ```typescript |
| 432 | * import { expect } from "alchemy-test"; |
| 433 | * import * as Cloudflare from "alchemy/Cloudflare"; |
| 434 | * import * as Test from "alchemy/Test/Alchemy"; |
| 435 | * import * as Effect from "effect/Effect"; |
| 436 | * import * as Layer from "effect/Layer"; |
| 437 | * import * as Schedule from "effect/Schedule"; |
| 438 | * import * as FetchHttpClient from "effect/http/FetchHttpClient"; |
| 439 | * import * as RpcClient from "effect/rpc/RpcClient"; |
| 440 | * import * as RpcSerialization from "effect/rpc/RpcSerialization"; |
| 441 | * import Stack from "../alchemy.run.ts"; |
| 442 | * import { TaskRpcs } from "../src/rpcs.ts"; |
| 443 | * |
| 444 | * const { test, beforeAll, afterAll, deploy, destroy } = Test.make({ |
| 445 | * providers: Cloudflare.providers(), |
| 446 | * }); |
| 447 | * const stack = beforeAll(deploy(Stack)); |
| 448 | * afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack)); |
| 449 | * |
| 450 | * const layer = (url: string) => |
| 451 | * RpcClient.layerProtocolHttp({ url }).pipe( |
| 452 | * Layer.provide(FetchHttpClient.layer), |
| 453 | * Layer.provide( |
| 454 | * Layer.succeed(RpcSerialization.RpcSerialization, RpcSerialization.json), |
| 455 | * ), |
| 456 | * ); |
| 457 | * |
| 458 | * test( |
| 459 | * "getTask", |
| 460 | * Effect.gen(function* () { |
| 461 | * const { url } = yield* stack; |
| 462 | * yield* Effect.gen(function* () { |
| 463 | * const client = yield* RpcClient.make(TaskRpcs); |
| 464 | * const result = yield* client |
| 465 | * .getTask({ id: "abc" }) |
| 466 | * .pipe(Effect.retry({ schedule: Schedule.exponential("500 millis"), times: 5 })); |
| 467 | * expect(result).toBe("task-abc"); |
| 468 | * }).pipe(Effect.scoped, Effect.provide(layer(url))); |
| 469 | * }), |
| 470 | * ); |
| 471 | * ``` |
| 472 | * |
| 473 | * ### Yielding the surrounding worker from inside the impl |
| 474 | * **Example:** `yield* RpcWorker` inside the init effect |
| 475 | * Mirrors `yield* DurableObject` — yield the tag to access |
| 476 | * the surrounding worker. |
| 477 | * ```typescript |
| 478 | * Effect.gen(function* () { |
| 479 | * const self = yield* Cloudflare.RpcWorker; |
| 480 | * }); |
| 481 | * ``` |
| 482 | * |
| 483 | * @resource |
| 484 | * @product Workers |
| 485 | * @category Workers & Compute |
| 486 | */ |
| 487 | export const RpcWorker: RpcWorkerClass = (() => { |
| 488 | const fn = (...args: any[]) => { |
| 489 | // Class-form: zero args returns the inner builder. Inner-arg arity |
| 490 | // distinguishes modular (`(id, props)`, no impl — `static make(impl)` |
| 491 | // provides the runtime) from inline-impl (`(id, props, impl)`). |
| 492 | if (args.length === 0) { |
| 493 | return (...inner: any[]) => { |
| 494 | if (inner.length === 2) { |
| 495 | const [id, props] = inner as [string, RpcWorkerProps<any>]; |
| 496 | return buildModular(id, props); |
| 497 | } |
| 498 | const [id, props, impl] = inner as [ |
| 499 | string, |
| 500 | RpcWorkerProps<any>, |
| 501 | Effect.Effect<Effect.Effect<HttpEffect<any>>>, |
| 502 | ]; |
| 503 | return build(id, props, impl); |
| 504 | }; |
| 505 | } |
| 506 | // Bare form: `(name, props, impl)`. |
| 507 | const [id, props, impl] = args as [ |
| 508 | string, |
| 509 | RpcWorkerProps<any>, |
| 510 | Effect.Effect<Effect.Effect<HttpEffect<any>>>, |
| 511 | ]; |
| 512 | return build(id, props, impl); |
| 513 | }; |
| 514 | |
| 515 | const tagged = taggedFunction(RpcWorkerScope, fn as any) as any; |
| 516 | tagged.bind = bind; |
| 517 | return tagged; |
| 518 | })() as any; |
| 519 | |
| 520 | const wrapImpl = (impl: Effect.Effect<Effect.Effect<HttpEffect<any>>>) => |
| 521 | // The user's inner Effect resolves to `HttpEffect`; the underlying |
| 522 | // `Cloudflare.Worker` expects `{ fetch: HttpEffect }`. Box it. |
| 523 | Effect.map( |
| 524 | impl, |
| 525 | (fetch) => ({ fetch }) as unknown as { fetch: HttpEffect<any> }, |
| 526 | ); |
| 527 | |
| 528 | // `effectClass` does not copy statics from the underlying Worker. |
| 529 | const stampLogicalId = ( |
| 530 | klass: Record<symbol | string, unknown>, |
| 531 | source: { LogicalId?: unknown }, |
| 532 | id: string, |
| 533 | ) => { |
| 534 | klass.LogicalId = |
| 535 | typeof source.LogicalId === "string" ? source.LogicalId : id; |
| 536 | return klass; |
| 537 | }; |
| 538 | |
| 539 | const buildModular = (id: string, props: RpcWorkerProps<any>) => { |
| 540 | const { schema } = props; |
| 541 | // Delegate to `Cloudflare.Worker<Self>()(id, props)` (modular form) |
| 542 | // so we inherit its `static make(impl)` plumbing for free. We just |
| 543 | // wrap the user's HttpEffect-returning impl into the `{ fetch }` |
| 544 | // shape `Cloudflare.Worker` expects, and stash the rpc schema on |
| 545 | // the class so `RpcWorker.bind(WorkerClass)` can recover it. |
| 546 | const Underlying: any = (WorkerCtor as any)()(id); |
| 547 | // `Underlying` is itself an Effect (the no-impl Worker class), so hand it |
| 548 | // straight to `effectClass` rather than reaching for a removed `.asEffect()`. |
| 549 | const klass = class extends effectClass(Underlying as Effect.Effect<Worker>) { |
| 550 | static make = ( |
| 551 | props: InputProps<WorkerProps>, |
| 552 | impl: Effect.Effect<Effect.Effect<HttpEffect<any>>>, |
| 553 | ): Layer.Layer<any, never, any> => Underlying.make(props, wrapImpl(impl)); |
| 554 | } as unknown as Record<symbol | string, unknown>; |
| 555 | klass[SchemaSymbol] = schema; |
| 556 | return stampLogicalId(klass, Underlying, id); |
| 557 | }; |
| 558 | |
| 559 | const build = ( |
| 560 | id: string, |
| 561 | props: RpcWorkerProps<any>, |
| 562 | impl: Effect.Effect<Effect.Effect<HttpEffect<any>>>, |
| 563 | ) => { |
| 564 | const { schema, ...workerProps } = props; |
| 565 | |
| 566 | // Wrap the user's HttpEffect-returning Effect into the `{ fetch }` |
| 567 | // shape `Cloudflare.Worker` expects. |
| 568 | const wrappedImpl = Effect.map( |
| 569 | impl, |
| 570 | (fetch) => ({ fetch }) as unknown as { fetch: HttpEffect<any> }, |
| 571 | ); |
| 572 | |
| 573 | // Delegate registration / binding metadata to the underlying |
| 574 | // `Cloudflare.Worker`. The returned `effectClass`, when yielded, |
| 575 | // produces the standard `Worker` resource (so `worker.url`, |
| 576 | // `worker.workerName`, etc. all work as usual). |
| 577 | const underlying = (WorkerCtor as any)()(id, workerProps, wrappedImpl); |
| 578 | |
| 579 | // Re-wrap as our own `effectClass` so we can stash the rpc schema |
| 580 | // on the class for `RpcWorker.bind` to recover later. (Re-using |
| 581 | // `effectClass` here means `class X extends RpcWorker<X>()(...)` |
| 582 | // still works.) `underlying` is already an Effect, so pass it directly. |
| 583 | const klass = effectClass( |
| 584 | underlying as Effect.Effect<Worker>, |
| 585 | ) as unknown as Record<symbol | string, unknown>; |
| 586 | klass[SchemaSymbol] = schema; |
| 587 | return stampLogicalId(klass, underlying, id); |
| 588 | }; |