1import type * as cf from "@cloudflare/workers-types";
2import type { DurableObject as DurableObjectClass } from "cloudflare:workers";
3
4import * as Cause from "effect/Cause";
5import * as Context from "effect/Context";
6import * as Effect from "effect/Effect";
7import * as Exit from "effect/Exit";
8import * as Layer from "effect/Layer";
9import * as Scope from "effect/Scope";
10import * as Stream from "effect/Stream";
11
12import { HttpServerResponse } from "effect/http";
13import {
14 dispatchAlarmCallbacks,
15 initializeAlarmCallbacks,
16 makeDurableObjectCallbackFactory,
17} from "./AlarmCallback.ts";
18import { RuntimeContext } from "../../RuntimeContext.ts";
19import { buildEventTelemetry } from "../../TelemetryRuntime.ts";
20import type {
21 DurableObjectExport,
22 DurableObjectShape,
23} from "./DurableObject.ts";
24import {
25 DurableObjectState,
26 fromDurableObjectState,
27} from "./DurableObjectState.ts";
28import { isScopeEjected, makeRequestEffect } from "./HttpServer.ts";
29import { fromWebSocket } from "./WebSocket.ts";
30import { getWorkerExport, handleRpcExit } from "./WorkerBridge.ts";
31
32/**
33 * Create a DurableObjectBridge class that proxies RPC method calls through
34 * the Effect runtime, encoding success/fail/stream results as RPC envelopes.
35 *
36 * Accepts the `DurableObject` base class and a `getExport` resolver so the
37 * implementation lives in real TypeScript instead of a generated string template.
38 */
39export const makeDurableObjectBridge =
40 (
41 DurableObject: typeof DurableObjectClass,
42 {
43 entrypoint,
44 stack,
45 }: {
46 entrypoint: Effect.Effect<Record<string, any>>;
47 stack: {
48 name: string;
49 stage: string;
50 };
51 },
52 ) =>
53 (className: string) => {
54 // One isolate-lifetime layer build shared by every activation of this DO
55 // class: `build` memoizes the built context, so re-activations (including
56 // hibernatable WebSocket wakes, which re-run the constructor) reuse it
57 // instead of rebuilding the layer stack.
58 const { build } = getWorkerExport<DurableObjectExport>({
59 entrypoint,
60 stack,
61 exportName: className,
62 });
63
64 return class DurableObjectBridge extends DurableObject {
65 #state;
66 #instance;
67 constructor(state: cf.DurableObjectState, env: any) {
68 super(state as any, env);
69 this.#state = state;
70
71 this.#instance = state.blockConcurrencyWhile(() =>
72 build((promise) => void (state as any).waitUntil?.(promise)).then(
73 ({ context, runtimeContext, export: exported, telemetry }) => {
74 const { constructor, services } = exported;
75 const instanceRuntimeContext = {
76 ...runtimeContext,
77 makeCallback: makeDurableObjectCallbackFactory(this.#state),
78 };
79 const doContext = Layer.succeed(
80 DurableObjectState,
81 fromDurableObjectState(this.#state),
82 ).pipe(
83 Layer.provideMerge(Layer.succeedContext(services)),
84 Layer.provideMerge(Layer.succeedContext(context)),
85 );
86 return constructor.pipe(
87 Effect.provide(doContext),
88 Effect.flatMap((instance) =>
89 Effect.suspend(() => {
90 const seal = initializeAlarmCallbacks(this.#state);
91 const instanceContext = Layer.succeed(
92 RuntimeContext,
93 instanceRuntimeContext,
94 ).pipe(Layer.provideMerge(doContext));
95 return instance.pipe(
96 Effect.provide(instanceContext),
97 Effect.ensuring(Effect.sync(seal)),
98 );
99 }),
100 ),
101 Effect.map((instance) => ({
102 instance,
103 services: Context.add(
104 services,
105 RuntimeContext,
106 instanceRuntimeContext,
107 ),
108 context,
109 telemetry,
110 })),
111 Effect.runPromise,
112 );
113 },
114 ),
115 );
116
117 return new Proxy(this, {
118 get: (target, prop) => {
119 const bind = (f: any) =>
120 typeof f === "function" ? f.bind(target) : f;
121 if (typeof prop !== "string") return bind((target as any)[prop]);
122 if (prop in target) return bind((target as any)[prop]);
123 return async (...args: any[]) =>
124 this.#execute((instance) => {
125 const method = instance[prop as keyof DurableObjectShape];
126 if (typeof method === "function") {
127 const result = (method as any)(...args);
128 // Effects (including nested-RPC values built by
129 // `asEffectOrStream`, which are Effects *branded* as Streams)
130 // must be run as effects — their resolved value may itself be
131 // a `Stream`, which `handleRpcExit` then encodes. Only a
132 // *genuine* `Stream` (not an Effect) is lifted into the
133 // success channel so `handleRpcExit` encodes it directly.
134 return Effect.isEffect(result)
135 ? result
136 : Stream.isStream(result)
137 ? Effect.succeed(result)
138 : result;
139 } else if (Effect.isEffect(method)) {
140 return method;
141 } else {
142 return Effect.succeed(method);
143 }
144 }, handleRpcExit);
145 },
146 });
147 }
148
149 async #execute(
150 fn: (instance: DurableObjectShape) => Effect.Effect<any, any, any>,
151 onExit?: (
152 exit: Exit.Exit<any, any>,
153 scope: Scope.Closeable,
154 ) => Promise<any>,
155 ) {
156 const scope = Scope.makeUnsafe();
157
158 const { instance, services, context, telemetry } = await this.#instance;
159
160 return fn(instance)
161 .pipe(
162 Effect.provide(
163 Layer.mergeAll(
164 Layer.succeed(
165 DurableObjectState,
166 fromDurableObjectState(this.#state),
167 ),
168 Layer.succeed(Scope.Scope, scope),
169 // The configured telemetry exporters, attached to the *call*
170 // scope by `buildEventTelemetry` so buffered telemetry
171 // flushes when the scope closes into `waitUntil` below (the
172 // isolate scope never finalizes on workerd).
173 Layer.effectContext(
174 buildEventTelemetry(context, scope, telemetry()),
175 ),
176 ).pipe(
177 Layer.provideMerge(Layer.succeedContext(services)),
178 Layer.provideMerge(Layer.succeedContext(context)),
179 ),
180 ),
181 Effect.runPromiseExit,
182 )
183 .then((exit) =>
184 onExit
185 ? onExit(exit, scope)
186 : exit._tag === "Success"
187 ? Promise.resolve(exit.value)
188 : Promise.reject(Cause.squash(exit.cause)),
189 )
190 .finally(() =>
191 isScopeEjected(scope)
192 ? undefined
193 : this.ctx.waitUntil(
194 // Match WorkerBridge: yield one macrotask so the
195 // HttpMiddleware tracer's late span-end reaches the
196 // telemetry exporter before the scope's flush finalizer.
197 new Promise((resolve) => setTimeout(resolve, 0)).then(() =>
198 Effect.runPromise(Scope.close(scope, Exit.void)),
199 ),
200 ),
201 );
202 }
203
204 async fetch(request: Request): Promise<any> {
205 return this.#execute((instance) =>
206 instance.fetch
207 ? makeRequestEffect(request as any, instance.fetch)
208 : Effect.succeed(
209 HttpServerResponse.text("Not implemented", {
210 status: 404,
211 }),
212 ),
213 );
214 }
215
216 async alarm(alarmInfo?: cf.AlarmInvocationInfo) {
217 return this.#execute((instance) =>
218 dispatchAlarmCallbacks(
219 this.#state,
220 instance.alarm !== undefined,
221 ).pipe(
222 Effect.andThen(() => instance.alarm?.(alarmInfo) ?? Effect.void),
223 ),
224 );
225 }
226
227 async webSocketMessage(ws: WebSocket, message: string | ArrayBuffer) {
228 return this.#execute(
229 (instance) =>
230 instance.webSocketMessage?.(fromWebSocket(ws as any), message) ??
231 Effect.void,
232 );
233 }
234
235 async webSocketClose(
236 ws: WebSocket,
237 code: number,
238 reason: string,
239 wasClean: boolean,
240 ) {
241 return this.#execute(
242 (instance) =>
243 instance.webSocketClose?.(
244 fromWebSocket(ws as any),
245 code,
246 reason,
247 wasClean,
248 ) ?? Effect.void,
249 );
250 }
251
252 async webSocketError(ws: WebSocket, error: unknown) {
253 return this.#execute(
254 (instance) =>
255 instance.webSocketError?.(fromWebSocket(ws as any), error) ??
256 Effect.void,
257 );
258 }
259 } as any;
260 };