1import type * as cf from "@cloudflare/workers-types";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import type { DurableObject, WorkerEntrypoint } from "cloudflare:workers";
4import * as Cause from "effect/Cause";
5import * as ConfigProvider from "effect/ConfigProvider";
6import * as Context from "effect/Context";
7import * as Effect from "effect/Effect";
8import * as Exit from "effect/Exit";
9import * as Layer from "effect/Layer";
10import * as Logger from "effect/Logger";
11import { MinimumLogLevel } from "effect/References";
12import * as Scope from "effect/Scope";
13import * as Stream from "effect/Stream";
14import * as FetchHttpClient from "effect/http/FetchHttpClient";
15import * as EffectHttp from "effect/http/HttpEffect";
16import {
17 makeEntrypointLayer,
18 reifyBoundConfigProvider,
19} from "../../Runtime.ts";
20import { RuntimeContext } from "../../RuntimeContext.ts";
21import { Self } from "../../Self.ts";
22import { StackContext } from "../../StackContext.ts";
23import { buildEventTelemetry } from "../../TelemetryRuntime.ts";
24import { CloudflareEnvironment } from "../CloudflareEnvironmentService.ts";
25import cloudflare_workers from "./cloudflare_workers.ts";
26import { isScopeEjected } from "./HttpServer.ts";
27import {
28 ErrorTag,
29 type RpcErrorEnvelope,
30 type RpcStreamEnvelope,
31 encodeRpcError,
32 toRpcStream,
33} from "./Rpc.ts";
34import {
35 ExportedHandlerMethods,
36 WorkerEnvironment,
37 WorkerExecutionContext,
38 deferredExecutionContext,
39 fromExecutionContext,
40} from "./WorkerRuntime.ts";
41import type { Worker } from "./Worker.ts";
42import type { WorkerRuntimeContext } from "./WorkerRuntimeContext.ts";
43
44/**
45 * The isolate-lifetime artifacts produced by a single layer build: the built
46 * service Context, the resolved export for this entrypoint, the user's
47 * RPC shape (a thunk — the shape is only populated once `serve` has run),
48 * and the telemetry Layer override registered during init (a thunk for the
49 * same reason).
50 */
51export interface WorkerBuild<Export = any> {
52 readonly context: Context.Context<any>;
53 readonly runtimeContext: WorkerRuntimeContext;
54 readonly export: Export;
55 readonly shape: () => Record<string, any>;
56 readonly telemetry: () => Layer.Layer<never, any, any> | undefined;
57}
58
59/**
60 * Makes the WorkerEntrypoint class and bridges to Effect fetch and RPC calls.
61 */
62export const makeWorkerBridge = (
63 Base: typeof WorkerEntrypoint | typeof DurableObject,
64 {
65 stack,
66 entrypoint,
67 }: {
68 stack: {
69 name: string;
70 stage: string;
71 };
72 entrypoint: any;
73 },
74) => {
75 const { build } = getWorkerExport({
76 entrypoint,
77 stack,
78 exportName: "default",
79 });
80
81 const processEvent = <T>(
82 makeEffect: (
83 build: WorkerBuild,
84 ) => readonly [Effect.Effect<any, any, any>, Context.Context<never>],
85 ctx: cf.ExecutionContext,
86 env: Record<string, unknown> | undefined,
87 onExit: (
88 exit: Exit.Exit<any, any>,
89 scope: Scope.Closeable,
90 ) => T | Promise<T>,
91 ): Promise<T> => {
92 const scope = Scope.makeUnsafe();
93 return build((promise) => ctx.waitUntil(promise as Promise<any>))
94 .then(
95 (built) => {
96 const [eff, services] = makeEffect(built);
97 return eff.pipe(
98 // Per-event services take precedence over the captured services
99 // and the built isolate context: the isolate context carries the
100 // *deferred* WorkerExecutionContext (yieldable in the top-level
101 // closure), which must be shadowed by the real per-event one
102 // here, and the fresh request `Scope` so `Effect.addFinalizer`
103 // in a handler attaches to the request scope (closed into
104 // `ctx.waitUntil` below).
105 Effect.provide(
106 Layer.mergeAll(
107 Layer.succeed(
108 WorkerExecutionContext,
109 fromExecutionContext(ctx, env),
110 ),
111 Layer.succeed(Scope.Scope, scope),
112 Layer.succeed(RuntimeContext, built.runtimeContext),
113 // The configured telemetry exporters. Constructed as part
114 // of this per-event layer, but `buildEventTelemetry`
115 // attaches their batching fibers and flush finalizers to
116 // the request `scope` (not this build's transient scope),
117 // so buffered telemetry flushes when the scope closes into
118 // `ctx.waitUntil` below — never on workerd's ephemeral
119 // isolate scope.
120 Layer.effectContext(
121 buildEventTelemetry(built.context, scope, built.telemetry()),
122 ),
123 ).pipe(
124 Layer.provideMerge(Layer.succeedContext(services)),
125 Layer.provideMerge(Layer.succeedContext(built.context)),
126 ),
127 ),
128 Effect.runPromiseExit,
129 );
130 },
131 // A failed isolate build reaches callers as a defect exit so the RPC
132 // path can envelope-encode it like any other handler defect.
133 (error) => Exit.die(error),
134 )
135 .then((exit) => onExit(exit, scope))
136 .finally(() =>
137 isScopeEjected(scope)
138 ? undefined
139 : ctx.waitUntil(
140 // The HttpMiddleware tracer ends the request's root span in a
141 // dispatcher task scheduled after the handler effect resolves.
142 // Yield one macrotask before closing the scope so that span
143 // reaches the telemetry exporter's buffer before the scope's
144 // flush finalizer runs.
145 new Promise((resolve) => setTimeout(resolve, 0)).then(() =>
146 Effect.runPromise(Scope.close(scope, Exit.void)),
147 ),
148 ),
149 );
150 };
151 class WorkerBridge extends Base {
152 constructor(
153 public readonly ctx: any,
154 public readonly env: any,
155 ) {
156 super(ctx, env);
157
158 for (const methodName of ExportedHandlerMethods) {
159 (this as any)[methodName] = async (input: any) =>
160 processEvent(
161 (built) =>
162 built.export[methodName](input, this.env, this.ctx) as [
163 Effect.Effect<any>,
164 Context.Context<never>,
165 ],
166 this.ctx,
167 this.env,
168 (exit) =>
169 exit._tag === "Success"
170 ? Promise.resolve(exit.value)
171 : Promise.reject(Cause.squash(exit.cause)),
172 );
173 }
174
175 return new Proxy(this, {
176 get: (target, prop) => {
177 if (typeof prop !== "string") return (target as any)[prop];
178 if (prop in target) return (target as any)[prop];
179 return (...args: unknown[]) =>
180 processEvent(
181 (built) => {
182 const dispatcher = built.shape()?.[prop];
183 if (typeof dispatcher !== "function") {
184 return [
185 Effect.die(
186 new Error(
187 `Method "${prop}" not found on worker. ` +
188 `Make sure it's returned from the worker's default export.`,
189 ),
190 ),
191 Context.empty(),
192 ] as const;
193 }
194 const result = dispatcher(...args);
195 // Effects (including nested-RPC values built by
196 // `asEffectOrStream`, which are Effects *branded* as Streams)
197 // must be run as effects — their resolved value may itself be
198 // a `Stream`, which `handleRpcExit` then encodes. Only a
199 // *genuine* `Stream` (not an Effect) is lifted into the
200 // success channel so `handleRpcExit` encodes it directly.
201 return [
202 Effect.isEffect(result)
203 ? (result as Effect.Effect<any>)
204 : Stream.isStream(result)
205 ? Effect.succeed(result)
206 : (result as Effect.Effect<any>),
207 Context.empty(),
208 ] as const;
209 },
210 this.ctx,
211 this.env,
212 handleRpcExit,
213 );
214 },
215 });
216 }
217 }
218
219 // Stub prototype methods so Cloudflare's script-validate detects the
220 // standard handler set; per-instance overrides above are what actually
221 // run.
222 for (const method of ExportedHandlerMethods) {
223 Object.defineProperty(WorkerBridge.prototype, method, {
224 value: function () {
225 throw new Error(
226 `Bridge method '${method}' was called before instance setup`,
227 );
228 },
229 writable: true,
230 configurable: true,
231 });
232 }
233
234 return WorkerBridge;
235};
236
237/**
238 * One isolate-lifetime layer build per entrypoint module. The generated
239 * entry passes the same `meta.entrypoint` object to `makeWorkerBridge`,
240 * `makeDurableObjectBridge`, and `makeWorkflowBridge`, so keying on it
241 * shares a single build (one run of the user's init closure) across the
242 * default worker and every Durable Object / Workflow class in the isolate.
243 */
244const sharedBuilds = new WeakMap<
245 object,
246 (pin: (promise: Promise<unknown>) => unknown) => Promise<Context.Context<any>>
247>();
248
249const getSharedBuild = (
250 entrypoint: any,
251 stack: { name: string; stage: string },
252) => {
253 let shared = sharedBuilds.get(entrypoint);
254 if (shared !== undefined) {
255 return shared;
256 }
257
258 const tag = Self as any as Context.Service<
259 never,
260 Worker & {
261 RuntimeContext: WorkerRuntimeContext;
262 }
263 >;
264
265 const layer = makeEntrypointLayer(tag, entrypoint);
266
267 const platform = Layer.mergeAll(
268 NodeServices.layer,
269 FetchHttpClient.layer,
270 // TODO(sam): wire this up to telemetry more directly
271 Logger.layer([Logger.consolePretty()]),
272 );
273
274 // Private scope for the isolate-lifetime layer build. Never closed —
275 // workerd has no isolate-teardown hook, so finalizers attached here can
276 // never run. It exists only because `Layer.buildWithMemoMap` requires a
277 // scope argument (`Layer.scoped`-style layers attach their finalizers to
278 // it). It is deliberately NOT provided as the ambient `Scope.Scope` of the
279 // init context: request-coupled resources are acquired inside handlers
280 // against the per-event scope that `processEvent` provides.
281 const instanceScope = Scope.makeUnsafe();
282 const memoMap = Layer.makeMemoMapUnsafe();
283
284 const globalContext = Layer.unwrap(
285 cloudflare_workers.pipe(
286 Effect.map(({ env }) =>
287 layer.pipe(
288 Layer.provideMerge(
289 Layer.succeed(StackContext, {
290 name: stack.name,
291 stage: stack.stage,
292 bindings: {},
293 resources: {},
294 actions: {},
295 }),
296 ),
297 Layer.provideMerge(platform),
298 Layer.provideMerge(
299 Layer.succeed(
300 ConfigProvider.ConfigProvider,
301 ConfigProvider.orElse(
302 ConfigProvider.fromUnknown({ ALCHEMY_PHASE: "runtime" }),
303 // Auto-bound `Config` values arrive in `env` as
304 // `{"_tag":"Redacted","value":...}` markers; reify them so a
305 // `Config` re-read inside a request handler decodes the raw
306 // source value instead of the marker JSON.
307 reifyBoundConfigProvider(
308 ConfigProvider.fromUnknown(env),
309 env as Record<string, unknown>,
310 ),
311 ),
312 ),
313 ),
314 Layer.provideMerge(Layer.succeed(WorkerEnvironment, env)),
315 // Init-phase ExecutionContext: yieldable from the Worker's
316 // top-level closure (and Layers); its RuntimeContext-colored
317 // methods defer to the real per-event context provided by
318 // `processEvent`.
319 Layer.provideMerge(
320 Layer.succeed(WorkerExecutionContext, deferredExecutionContext),
321 ),
322 Layer.provideMerge(
323 Layer.succeed(
324 CloudflareEnvironment,
325 // TODO(sam): fix this with maybe a CloudflareAccountId Effect service
326 // @ts-expect-error - this is hacky, but we only need and have this property
327 Effect.succeed({
328 account: (env as any).ALCHEMY_CLOUDFLARE_ACCOUNT_ID,
329 }),
330 ),
331 ),
332 Layer.provideMerge(
333 Layer.succeed(
334 MinimumLogLevel,
335 (env as any).DEBUG ? "Debug" : "Info",
336 ),
337 ),
338 ),
339 ),
340 ),
341 );
342
343 let built: Promise<Context.Context<any>> | undefined;
344
345 /**
346 * Build the isolate-lifetime layer stack exactly once; every subsequent
347 * event (and every export sharing this entrypoint) reuses the memoized
348 * Context.
349 *
350 * `pin` registers the in-flight build promise with the calling event
351 * (`ctx.waitUntil` / `state.waitUntil`). Every awaiting event must pin:
352 * workerd schedules a promise's continuations back into its origin request
353 * context and *drops* them if that context has ended
354 * (`handle_cross_request_promise_resolution`), so the origin event must be
355 * kept alive until the build settles or concurrent cold-start requests
356 * would hang.
357 *
358 * Only success is memoized — a transient init failure (e.g. a flaky
359 * `Config` read in user init) resets the memo and heals on the next event.
360 */
361 shared = (pin) => {
362 const promise = (built ??= Effect.runPromise(
363 Layer.buildWithMemoMap(globalContext, memoMap, instanceScope).pipe(
364 // Strip the build's memo map from the exposed context so a Layer the
365 // user `Effect.provide`s *inside a handler* builds per event instead
366 // of sharing one instance (pinned to the first request's IoContext)
367 // across concurrent events.
368 Effect.map(Context.omit(Layer.CurrentMemoMap)),
369 ),
370 ).catch((error) => {
371 built = undefined;
372 throw error;
373 }));
374 pin(promise.catch(() => {}));
375 return promise;
376 };
377 sharedBuilds.set(entrypoint, shared);
378 return shared;
379};
380
381export const getWorkerExport = <Export = any>({
382 entrypoint,
383 stack,
384 exportName,
385}: {
386 entrypoint: any;
387 stack: { name: string; stage: string };
388 exportName: string;
389}) => {
390 const tag = Self as any as Context.Service<
391 never,
392 Worker & {
393 RuntimeContext: WorkerRuntimeContext;
394 }
395 >;
396
397 const runtimeContext = tag.pipe(Effect.map((func) => func.RuntimeContext));
398 const exported = runtimeContext.pipe(
399 Effect.flatMap((context) => context.exports),
400 Effect.flatMap((exports) =>
401 Effect.isEffect(exports[exportName])
402 ? exports[exportName]
403 : Effect.succeed(exports[exportName]),
404 ),
405 ) as Effect.Effect<Export>;
406
407 const sharedBuild = getSharedBuild(entrypoint, stack);
408
409 let built: Promise<WorkerBuild<Export>> | undefined;
410
411 /**
412 * Resolve this export against the shared isolate build; memoized so
413 * listener assembly and the captured services context resolve once per
414 * export. Same success-only memoization contract as the shared build.
415 */
416 const build = (
417 pin: (promise: Promise<unknown>) => unknown,
418 ): Promise<WorkerBuild<Export>> => {
419 const promise = (built ??= sharedBuild(pin)
420 .then((context) =>
421 Effect.runPromise(
422 Effect.all([exported, runtimeContext]).pipe(
423 Effect.map(([exp, rc]): WorkerBuild<Export> => ({
424 context,
425 runtimeContext: rc,
426 export: exp,
427 shape: rc.shape,
428 telemetry: () => rc.telemetry,
429 })),
430 Effect.provideContext(context),
431 ),
432 ),
433 )
434 .catch((error) => {
435 built = undefined;
436 throw error;
437 }));
438 pin(promise.catch(() => {}));
439 return promise;
440 };
441
442 return { build };
443};
444
445export const makeRpcProxy = (
446 self: any,
447 userShape: Effect.Effect<any>,
448 processEvent: (
449 eff: Effect.Effect<[Effect.Effect<any>, Context.Context<never>]>,
450 ) => Promise<any>,
451) =>
452 new Proxy(self, {
453 get: (target, prop) => {
454 if (typeof prop !== "string") return (target as any)[prop];
455 if (prop in target) return (target as any)[prop];
456 return (...args: unknown[]) =>
457 userShape
458 .pipe(
459 Effect.map((shape) => shape[prop]),
460 Effect.flatMap((dispatcher) => {
461 if (typeof dispatcher !== "function") {
462 return Effect.die(
463 new Error(
464 `Method "${prop}" not found on worker. ` +
465 `Make sure it's returned from the worker's default export.`,
466 ),
467 );
468 }
469 const result = dispatcher(...args);
470 // Effects (including nested-RPC values built by
471 // `asEffectOrStream`, which are Effects *branded* as Streams)
472 // must be run as effects — their resolved value may itself be a
473 // `Stream`, which `handleRpcExit` then encodes. Only a *genuine*
474 // `Stream` (not an Effect) is lifted into the success channel so
475 // `handleRpcExit` encodes it directly.
476 return Effect.isEffect(result)
477 ? (result as Effect.Effect<any>)
478 : Stream.isStream(result)
479 ? Effect.succeed(result)
480 : (result as Effect.Effect<any>);
481 }),
482 processEvent,
483 )
484 .then((exit) => handleRpcExit(exit));
485 },
486 });
487
488export const handleRpcExit = async (
489 exit: Exit.Exit<any, any>,
490 scope?: Scope.Closeable,
491) => {
492 if (exit._tag === "Success") {
493 if (Stream.isStream(exit.value)) {
494 let stream = exit.value as Stream.Stream<any, any, any>;
495 if (scope !== undefined && !isScopeEjected(scope)) {
496 // The RPC transport drains the encoded ReadableStream *after* this
497 // function returns, so the request scope must outlive the handler:
498 // eject it from the bridge's close-on-return path and close it when
499 // the stream settles instead — mirroring `scopeTransferToStream` on
500 // the fetch path.
501 EffectHttp.scopeDisableClose(scope);
502 stream = stream.pipe(
503 Stream.onExit((streamExit) => Scope.close(scope, streamExit)),
504 );
505 }
506 return await Effect.runPromise(
507 toRpcStream(stream) as Effect.Effect<RpcStreamEnvelope>,
508 );
509 }
510 return exit.value;
511 }
512 const failReason = exit.cause.reasons.find(Cause.isFailReason);
513 if (failReason) {
514 return {
515 _tag: ErrorTag,
516 error: encodeRpcError(failReason.error),
517 } satisfies RpcErrorEnvelope;
518 }
519 const dieReason = exit.cause.reasons.find(Cause.isDieReason);
520 throw (
521 dieReason?.defect ?? new Error("RPC method failed with an unexpected cause")
522 );
523};