1import type * as cf from "@cloudflare/workers-types";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Option from "effect/Option";
5import type { HttpEffect } from "../../Http.ts";
6import * as Output from "../../Output.ts";
7import {
8 packEnvValueKeepRedacted,
9 unpackEnvValue,
10} from "../../RuntimeContext.ts";
11import type * as Serverless from "../../Serverless/index.ts";
12import type { DurableObjectExport } from "./DurableObject.ts";
13import { makeRequestHandler } from "./HttpServer.ts";
14import {
15 ExportedHandlerMethods,
16 WorkerEnvironment,
17 WorkerExecutionContext,
18 WorkerTypeId,
19 deferredExecutionContext,
20 type WorkerEvent,
21} from "./WorkerRuntime.ts";
22import type { WorkflowExport } from "../Workflows/Workflow.ts";
23import type { SqlMigrationsExport } from "./SqlMigrationsRuntime.ts";
24
25export type WorkerExport =
26 | DurableObjectExport
27 | WorkflowExport
28 | SqlMigrationsExport;
29
30export interface WorkerRuntimeContext extends Serverless.FunctionContext {
31 export(name: string, value: any): Effect.Effect<void>;
32 shape: () => Record<string, any>;
33}
34
35export const makeWorkerRuntimeContext = (id: string): WorkerRuntimeContext => {
36 const listeners: Effect.Effect<Serverless.FunctionListener>[] = [];
37 const exports: Record<string, WorkerExport> = {};
38 const env: Record<string, any> = {};
39 let userShape: Record<string, unknown> | undefined;
40
41 const ctx = {
42 Type: WorkerTypeId,
43 id,
44 env,
45 shape: () => userShape!,
46 get: (key: string) =>
47 Effect.serviceOption(WorkerEnvironment).pipe(
48 Effect.map(Option.getOrUndefined),
49 // Key is already canonical (see RuntimeContext.sanitizeKey). Read
50 // straight from `WorkerEnvironment` — see `unpackEnvValue` for why
51 // this must never resolve through `Config.String`.
52 Effect.map((env) => unpackEnvValue(env?.[key])),
53 ) as any,
54 set: (key: string, output: Output.Output) =>
55 Effect.sync(() => {
56 // `packEnvValueKeepRedacted` keeps the Redacted wrapper on the
57 // outside so the put-worker loop can deploy secrets via
58 // `secret_text` instead of leaking them as `plain_text`, while the
59 // inner marker lets the runtime `get` accessor rebuild the wrapper
60 // after Cloudflare hands the binding back as a plain string.
61 env[key] = output.pipe(Output.map(packEnvValueKeepRedacted));
62 return key;
63 }),
64 serve: <Req = never>(
65 handler: HttpEffect<Req> | Effect.Effect<HttpEffect<Req>>,
66 options?: { shape?: Record<string, unknown> },
67 ) => {
68 // Capture the user's full default-export shape so `exports` can
69 // expose any non-handler methods on it as RPC methods on the
70 // deployed `WorkerEntrypoint` subclass — see `__rpc__` below.
71 if (options?.shape) userShape = options.shape;
72 return ctx.listen(makeRequestHandler(handler));
73 },
74 listen: ((
75 handler:
76 | Serverless.FunctionListener
77 | Effect.Effect<Serverless.FunctionListener>,
78 ) =>
79 Effect.sync(() =>
80 Effect.isEffect(handler)
81 ? listeners.push(handler)
82 : listeners.push(Effect.succeed(handler)),
83 )) as any as Serverless.FunctionContext["listen"],
84 export: (name: string, value: any) =>
85 Effect.sync(() => {
86 exports[name] = value;
87 }),
88 planServices: Layer.mergeAll(
89 Layer.succeed(WorkerEnvironment, {}),
90 // Lets the init closure `yield*` WorkerExecutionContext during plan;
91 // its RuntimeContext-colored methods can't run until a real handler
92 // provides the live per-event context.
93 Layer.succeed(WorkerExecutionContext, deferredExecutionContext),
94 ),
95 exports: Effect.gen(function* () {
96 const handlers = yield* Effect.all(listeners, {
97 concurrency: "unbounded",
98 });
99 const services = yield* Effect.context();
100
101 const dispatch =
102 (type: WorkerEvent["type"]) =>
103 (request: any, env: unknown, context: cf.ExecutionContext) => {
104 const event: WorkerEvent = {
105 kind: "Cloudflare.Workers.WorkerEvent",
106 type,
107 input: request,
108 env,
109 context,
110 };
111 const effects: Effect.Effect<unknown>[] = [];
112 for (const handler of handlers) {
113 const eff = handler(event);
114 if (Effect.isEffect(eff)) {
115 effects.push(eff);
116 }
117 }
118 if (effects.length === 1) {
119 return [effects[0], services];
120 }
121 if (effects.length > 1) {
122 return [
123 Effect.all(effects, { concurrency: "unbounded" }).pipe(
124 Effect.map((results) => {
125 for (const result of results) {
126 if (result instanceof Response) return result;
127 }
128 return results[results.length - 1];
129 }),
130 ),
131 services,
132 ];
133 }
134 return [
135 Effect.die(
136 new Error(`No event handler found for event type '${type}'`),
137 ),
138 services,
139 ];
140 };
141
142 // RPC method dispatchers — one per non-handler method on the user's
143 // shape. Each dispatcher is invoked by the WorkerEntrypoint bridge
144 // as `dispatcher(args, ctx)`: `args` are the user-facing call args,
145 // `ctx` is the `this.ctx` that Cloudflare hands the bridge per RPC
146 // request. The dispatcher runs the user effect with the same runtime
147 // layer the fetch path uses, then envelope-encodes the result so
148 // `Effect.fail` round-trips as `RpcErrorEnvelope` and `Stream` as
149 // `RpcStreamEnvelope` (consumers wrap the binding with
150 // `toRpcAsync`/`bindWorker` to decode).
151
152 return {
153 ...exports,
154 default: Object.fromEntries(
155 ExportedHandlerMethods.map((method) => [method, dispatch(method)]),
156 ),
157 };
158 }),
159 };
160 return ctx;
161};