1import {
2 CurrentActivity,
3 decodeExit,
4 decodeResult,
5 encodeExit,
6 encodeResult,
7 makeCurrentActivity,
8 payloadCodec,
9} from "alchemy/WorkflowEncoding";
10import * as Cause from "effect/Cause";
11import * as Context from "effect/Context";
12import * as Duration from "effect/Duration";
13import * as Effect from "effect/Effect";
14import * as Exit from "effect/Exit";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import * as Schedule from "effect/Schedule";
18import * as Schema from "effect/Schema";
19import { Activity, Workflow as EffectWorkflow, WorkflowEngine } from "effect/workflow";
20import type { Worker } from "../Workers/Worker.ts";
21import { Workflow, type WorkflowHandle } from "./Workflow.ts";
22import { WorkflowStep } from "./WorkflowRuntime.ts";
23
24/** What each Cloudflare Workflow instance runs: one execution of one Effect workflow. */
25interface Invocation {
26 readonly workflow: string;
27 readonly executionId: string;
28 readonly payload: unknown;
29}
30
31type Execute = (
32 payload: object,
33 executionId: string,
34) => Effect.Effect<
35 unknown,
36 unknown,
37 WorkflowEngine.WorkflowInstance | WorkflowEngine.WorkflowEngine
38>;
39
40/** The run in progress: its step context, and the clocks it has scheduled. */
41class CurrentRun extends Context.Service<
42 CurrentRun,
43 {
44 readonly step: WorkflowStep["Service"];
45 readonly context: Context.Context<never>;
46 readonly clocks: Map<string, Duration.Duration>;
47 readonly instance: WorkflowEngine.WorkflowInstance["Service"];
48 }
49>()("alchemy/Cloudflare/Workflows/NativeEngine/CurrentRun") {}
50
51const currentRun = (operation: string) =>
52 Effect.flatMap(Effect.serviceOption(CurrentRun), (run) =>
53 Option.isSome(run)
54 ? Effect.succeed(run.value)
55 : Effect.die(new Error(`NativeEngine: ${operation} only runs inside a workflow run`)),
56 );
57
58/**
59 * A name Cloudflare accepts for an instance id or event type: up to 100
60 * letters, digits, `-` and `_`.
61 */
62const nameOf = (...parts: ReadonlyArray<string>) => {
63 const name = parts.join("-").replace(/[^A-Za-z0-9_-]/g, "_");
64 if (name.length <= 100) return name;
65 let hash = 2166136261;
66 for (const char of name) hash = Math.imul(hash ^ char.charCodeAt(0), 16777619) >>> 0;
67 return `${name.slice(0, 91)}-${hash.toString(16).padStart(8, "0")}`;
68};
69
70/** Every wait for a signal also waits for this event, which interrupts the run. */
71const INTERRUPT = "alchemy-interrupt";
72
73/** A run suspended on a failure waits for this event, which resumes it. */
74const RESUME = "alchemy-resume";
75
76/** A defect the workflow doesn't capture, failing its activity's step so Cloudflare retries it. */
77interface ActivityDefect {
78 readonly _tag: "ActivityDefect";
79 readonly message: string;
80}
81
82/** What an activity's step records when the activity suspends on a failure. */
83const SUSPENDED = "alchemy-suspended";
84
85/** How long a run waits for a signal; Cloudflare's limit for `waitForEvent`. */
86const SIGNAL_TIMEOUT = "365 days";
87
88/**
89 * Retry a Workflows binding call while the Workflow class is still
90 * propagating after a deploy, which Cloudflare reports as "Worker not found",
91 * for about a minute.
92 */
93const inTransit = <A>(call: Effect.Effect<A>): Effect.Effect<A> =>
94 call.pipe(
95 Effect.catchDefect((defect) =>
96 String(defect).includes("Worker not found") ? Effect.fail(defect) : Effect.die(defect),
97 ),
98 Effect.retry({
99 schedule: Schedule.min([Schedule.exponential("100 millis"), Schedule.spaced("2 seconds")]),
100 times: 35,
101 }),
102 Effect.orDie,
103 ) as Effect.Effect<A>;
104
105/**
106 * Send an event to an instance, retrying any failure for about a minute: a run
107 * takes each signal and interrupt at most once, so sending one again is safe.
108 * A finished instance takes no more events, and needs none.
109 */
110const sendEvent = (
111 handle: WorkflowHandle,
112 id: string,
113 event: { type: string; payload?: unknown },
114) =>
115 handle.get(id).pipe(
116 Effect.flatMap((instance) => instance.sendEvent(event)),
117 Effect.catchDefect((defect) => Effect.fail(defect)),
118 Effect.catch((error) =>
119 Effect.flatMap(statusOf(handle, id), (status) =>
120 status?.status === "complete" ||
121 status?.status === "errored" ||
122 status?.status === "terminated"
123 ? Effect.void
124 : Effect.fail(error),
125 ),
126 ),
127 Effect.retry({
128 schedule: Schedule.min([Schedule.exponential("100 millis"), Schedule.spaced("2 seconds")]),
129 times: 35,
130 }),
131 Effect.orDie,
132 ) as Effect.Effect<void>;
133
134/** An instance's status, or undefined if there's no such instance. */
135const statusOf = (handle: WorkflowHandle, id: string) =>
136 inTransit(handle.get(id).pipe(Effect.flatMap((instance) => instance.status()))).pipe(
137 Effect.exit,
138 Effect.map((exit) => (Exit.isSuccess(exit) ? exit.value : undefined)),
139 );
140
141/** An instance's result, or undefined while it runs. */
142const resultOf = (
143 handle: WorkflowHandle,
144 workflow: EffectWorkflow.AnyWithProps,
145 id: string,
146): Effect.Effect<EffectWorkflow.Complete<unknown, unknown> | undefined | null> =>
147 Effect.gen(function* () {
148 const status = yield* statusOf(handle, id);
149 if (status === undefined) return null;
150 switch (status.status) {
151 case "complete":
152 return yield* decodeResult(workflow, status.output as string);
153 case "terminated":
154 return new EffectWorkflow.Complete({ exit: Exit.interrupt() });
155 case "errored":
156 return new EffectWorkflow.Complete({
157 exit: Exit.die(new Error(status.error?.message ?? `Workflow instance ${id} errored`)),
158 });
159 default:
160 return undefined;
161 }
162 });
163
164/**
165 * Run Effect workflows on Cloudflare Workflows. One Cloudflare Workflow class,
166 * `NativeEngine`, runs every workflow the Worker implements; each execution is
167 * one instance, named after the workflow and its execution id.
168 *
169 * Steps run with `step.do`, so a finished step is replayed rather than run
170 * again. `DurableClock.sleep` is `step.sleep`, and `DurableDeferred.await`
171 * waits for an event that a run's `signal` sends.
172 *
173 * ### Running workflows on Cloudflare Workflows
174 * **Example:** Provide the engine to a Worker
175 * ```typescript
176 * import * as Cloudflare from "@alchemy.run/cloudflare";
177 * import * as Workflows from "alchemy/Workflows";
178 * import * as Effect from "effect/Effect";
179 * import * as Layer from "effect/Layer";
180 * import * as HttpServerResponse from "effect/http/HttpServerResponse";
181 *
182 * export default class AppWorker extends Cloudflare.Worker<AppWorker>()(
183 * "AppWorker",
184 * { main: import.meta.url },
185 * Effect.gen(function* () {
186 * const welcome = yield* Workflows.RunWorkflow(SendWelcome);
187 * return {
188 * fetch: Effect.gen(function* () {
189 * const run = yield* welcome.start({ userId: "u-1" });
190 * return HttpServerResponse.text(run.executionId);
191 * }),
192 * };
193 * }).pipe(
194 * Effect.provide(SendWelcomeLive.pipe(Layer.provideMerge(Cloudflare.Workflows.NativeEngine))),
195 * ),
196 * ) {}
197 * ```
198 *
199 * @layer
200 * @provides WorkflowEngine
201 * @product Workflows
202 */
203export const NativeEngine: Layer.Layer<WorkflowEngine.WorkflowEngine, never, Worker> = Layer.effect(
204 WorkflowEngine.WorkflowEngine,
205 Effect.gen(function* () {
206 const handlers = new Map<
207 string,
208 { readonly workflow: EffectWorkflow.AnyWithProps; readonly execute: Execute }
209 >();
210 let engine: WorkflowEngine.WorkflowEngine["Service"];
211
212 const handle = yield* Workflow(
213 "NativeEngine",
214 Effect.succeed((invocation: Invocation) =>
215 Effect.gen(function* () {
216 const registration = handlers.get(invocation.workflow);
217 if (!registration) {
218 return yield* Effect.die(
219 new Error(`Workflow ${invocation.workflow} is not hosted here`),
220 );
221 }
222 const step = yield* WorkflowStep;
223 const context = yield* Effect.context<never>();
224 const { workflow, execute } = registration;
225 const payload = yield* Schema.decodeEffect(payloadCodec(workflow))(
226 invocation.payload,
227 ).pipe(Effect.orDie);
228 const instance = WorkflowEngine.WorkflowInstance.initial(
229 workflow,
230 invocation.executionId,
231 );
232 const result = yield* execute(payload as object, invocation.executionId).pipe(
233 EffectWorkflow.intoResult,
234 Effect.provide(
235 Layer.mergeAll(
236 Layer.succeed(WorkflowEngine.WorkflowInstance, instance),
237 Layer.succeed(WorkflowEngine.WorkflowEngine, engine),
238 Layer.succeed(CurrentRun, { step, context, clocks: new Map(), instance }),
239 ),
240 ),
241 );
242 // Every wait blocks inside the instance, so a run always finishes.
243 if (result._tag === "Suspended") {
244 return yield* Effect.die(new Error(`Workflow ${workflow._tag} suspended unexpectedly`));
245 }
246 return yield* encodeResult(workflow, result.exit);
247 }),
248 ),
249 ) as Effect.Effect<WorkflowHandle, never, Worker>;
250
251 engine = WorkflowEngine.makeUnsafe({
252 register: (workflow, execute) =>
253 Effect.sync(
254 () =>
255 void handlers.set(workflow._tag, {
256 workflow: workflow as EffectWorkflow.AnyWithProps,
257 execute,
258 }),
259 ),
260
261 execute: (workflow, opts) =>
262 Effect.gen(function* () {
263 const id = nameOf(workflow._tag, opts.executionId);
264 const payload = yield* Schema.encodeEffect(
265 payloadCodec(workflow as EffectWorkflow.AnyWithProps),
266 )(opts.payload).pipe(Effect.orDie);
267 // The same key always names the same run: create it only once.
268 if ((yield* statusOf(handle, id)) === undefined) {
269 yield* inTransit(
270 handle.create({
271 id,
272 params: { workflow: workflow._tag, executionId: opts.executionId, payload },
273 }),
274 ).pipe(
275 Effect.exit,
276 Effect.flatMap((exit) =>
277 Exit.isSuccess(exit)
278 ? Effect.void
279 : Effect.flatMap(statusOf(handle, id), (status) =>
280 status === undefined ? Effect.failCause(exit.cause) : Effect.void,
281 ),
282 ),
283 );
284 }
285 if (opts.discard) return undefined;
286 const run = yield* Effect.serviceOption(CurrentRun);
287 if (Option.isNone(run)) {
288 const result = yield* resultOf(handle, workflow as EffectWorkflow.AnyWithProps, id);
289 return result ?? EffectWorkflow.Suspended.make({});
290 }
291 // A child workflow: wait for it here, sleeping between checks.
292 for (let check = 0; ; check++) {
293 const text = yield* run.value.step.do({
294 name: `child/${id}/${check}`,
295 effect: resultOf(handle, workflow as EffectWorkflow.AnyWithProps, id).pipe(
296 Effect.flatMap((result) =>
297 result
298 ? encodeResult(workflow as EffectWorkflow.AnyWithProps, result.exit)
299 : Effect.succeed(""),
300 ),
301 ),
302 });
303 if (text !== "") {
304 return yield* decodeResult(workflow as EffectWorkflow.AnyWithProps, text);
305 }
306 yield* run.value.step.sleep(`child-wait/${id}/${check}`, "5 seconds");
307 }
308 }) as Effect.Effect<any>,
309
310 poll: (workflow, executionId) =>
311 Effect.map(
312 resultOf(
313 handle,
314 workflow as EffectWorkflow.AnyWithProps,
315 nameOf(workflow._tag, executionId),
316 ),
317 (result) =>
318 result === null
319 ? Option.none()
320 : Option.some(result ?? EffectWorkflow.Suspended.make({})),
321 ),
322
323 // The run wakes from whatever it waits for, fails there, and runs
324 // its compensations.
325 interrupt: (workflow, executionId) =>
326 sendEvent(handle, nameOf(workflow._tag, executionId), { type: INTERRUPT }),
327
328 interruptUnsafe: (workflow, executionId) =>
329 inTransit(
330 handle
331 .get(nameOf(workflow._tag, executionId))
332 .pipe(Effect.flatMap((instance) => instance.terminate())),
333 ),
334
335 // Only a run suspended on a failure waits to be resumed.
336 resume: (workflow, executionId) =>
337 sendEvent(handle, nameOf(workflow._tag, executionId), { type: RESUME }),
338
339 activityExecute: (activity, attempt) =>
340 Effect.gen(function* () {
341 const run = yield* currentRun("activity execution");
342 const instance = yield* WorkflowEngine.WorkflowInstance;
343 const activityInstance = WorkflowEngine.WorkflowInstance.initial(
344 instance.workflow,
345 instance.executionId,
346 );
347 activityInstance.interrupted = instance.interrupted;
348 // The step's exit, failure or not, is its result. A defect the
349 // workflow doesn't capture fails the step instead, so Cloudflare
350 // retries it, and is a defect again once retries run out. An
351 // activity that suspends on a failure records that, and a resume
352 // runs it again under a new step name.
353 for (let generation = 0; ; generation++) {
354 const text = yield* run.step
355 .do({
356 name:
357 generation === 0
358 ? `${activity.name}/${attempt}`
359 : `${activity.name}/${attempt}/resumed-${generation}`,
360 effect: activity.executeEncoded.pipe(
361 EffectWorkflow.intoResult,
362 Effect.catchCauseIf(
363 (cause) => !Cause.hasInterruptsOnly(cause),
364 (cause) =>
365 Effect.fail<ActivityDefect>({
366 _tag: "ActivityDefect",
367 message: Cause.pretty(cause),
368 }),
369 ),
370 Effect.flatMap((result) =>
371 result._tag === "Complete"
372 ? encodeExit(result.exit)
373 : Effect.succeed(SUSPENDED),
374 ),
375 // The run's services, with the step's own instance.
376 Effect.provide(
377 Layer.succeedContext(
378 run.context.pipe(
379 Context.add(WorkflowEngine.WorkflowInstance, activityInstance),
380 // The step runs on its own fiber: carry the attempt `Activity.retry` set.
381 Context.add(Activity.CurrentAttempt, attempt),
382 Context.add(
383 CurrentActivity,
384 makeCurrentActivity(
385 instance.workflow._tag,
386 instance.executionId,
387 activity.name,
388 ),
389 ),
390 ),
391 ),
392 ),
393 ) as Effect.Effect<string, ActivityDefect>,
394 })
395 .pipe(
396 Effect.catchTag("ActivityDefect", (defect) =>
397 Effect.die(new Error(defect.message)),
398 ),
399 );
400 if (text !== SUSPENDED) {
401 return new EffectWorkflow.Complete({ exit: yield* decodeExit(text) });
402 }
403 // Suspended on a failure: wait here for a resume, then run the
404 // activity again, or for an interrupt.
405 const waiting = `${activity.name}/${attempt}/suspended-${generation}`;
406 const resumed = yield* run.step
407 .waitForEvent(waiting, { type: RESUME, timeout: SIGNAL_TIMEOUT })
408 .pipe(
409 Effect.as(true),
410 Effect.raceFirst(
411 run.step
412 .waitForEvent(`${waiting}/interrupt`, {
413 type: INTERRUPT,
414 timeout: SIGNAL_TIMEOUT,
415 })
416 .pipe(Effect.as(false)),
417 ),
418 );
419 if (!resumed) {
420 run.instance.interrupted = true;
421 run.instance.suspended = false;
422 return yield* Effect.interrupt;
423 }
424 }
425 }),
426
427 deferredResult: (deferred) =>
428 Effect.gen(function* () {
429 const run = yield* currentRun("DurableDeferred reads");
430 // A race asks for its recorded winner before it runs. A step can only
431 // be replayed by running it, so a winner can't be looked up first.
432 if (deferred.name.startsWith("raceAll/")) {
433 return yield* Effect.die(
434 new Error(
435 "Activity.raceAll and DurableDeferred.raceAll aren't supported on Cloudflare Workflows: a step's result can't be looked up before the step runs. Use the Durable Object engine.",
436 ),
437 );
438 }
439 const duration = run.clocks.get(deferred.name);
440 // Every wait also waits for an interrupt.
441 const interrupt = run.step
442 .waitForEvent(`${deferred.name}/interrupt`, {
443 type: INTERRUPT,
444 timeout: SIGNAL_TIMEOUT,
445 })
446 .pipe(Effect.as({ interrupted: true, payload: "" }));
447 const signal = yield* (
448 duration !== undefined
449 ? run.step
450 .sleep(deferred.name, Duration.toMillis(duration))
451 .pipe(Effect.as({ interrupted: false, payload: "" }))
452 : run.step
453 .waitForEvent<string>(deferred.name, {
454 type: nameOf(deferred.name),
455 timeout: SIGNAL_TIMEOUT,
456 })
457 .pipe(Effect.map((event) => ({ interrupted: false, payload: event.payload })))
458 ).pipe(Effect.raceFirst(interrupt));
459 if (signal.interrupted) {
460 run.instance.interrupted = true;
461 run.instance.suspended = false;
462 return yield* Effect.interrupt;
463 }
464 return Option.some(
465 duration !== undefined ? Exit.void : yield* decodeExit(signal.payload),
466 );
467 }),
468
469 deferredDone: (opts) =>
470 Effect.gen(function* () {
471 const payload = yield* encodeExit(opts.exit);
472 yield* sendEvent(handle, nameOf(opts.workflowName, opts.executionId), {
473 type: nameOf(opts.deferredName),
474 payload,
475 });
476 }),
477
478 scheduleClock: (_workflow, opts) =>
479 Effect.flatMap(currentRun("DurableClock scheduling"), (run) =>
480 Effect.sync(() => void run.clocks.set(opts.clock.deferred.name, opts.clock.duration)),
481 ),
482 });
483 return engine;
484 }),
485);