1import * as Cause from "effect/Cause";
2import type * as Context from "effect/Context";
3import * as Effect from "effect/Effect";
4import { Base64, Base64Url } from "effect/encoding";
5import * as Exit from "effect/Exit";
6import * as FiberSet from "effect/FiberSet";
7import * as Layer from "effect/Layer";
8import * as Result from "effect/Result";
9import * as Scope from "effect/Scope";
10import { encodeRpcError } from "../../Rpc.ts";
11import { buildEventTelemetry } from "../../TelemetryRuntime.ts";
12import { isScopeEjected } from "../Workers/HttpServer.ts";
13import { getWorkerExport } from "../Workers/WorkerBridge.ts";
14import type {
15 WorkflowExport,
16 WorkflowImpl,
17 WorkflowStepConfig,
18 WorkflowStepEvent,
19 WorkflowTaskOptions,
20} from "./Workflow.ts";
21import {
22 WorkflowEvent as WorkflowEventService,
23 WorkflowStep,
24 WorkflowStepContext,
25} from "./WorkflowRuntime.ts";
26
27/**
28 * Create a WorkflowBridge class that extends `WorkflowEntrypoint` and
29 * delegates the `run(event, step)` call to the Effect-native workflow body
30 * registered via `worker.export(...)`.
31 *
32 * The bridge provides `WorkflowEvent` and `WorkflowStep` as Effect
33 * services so the user writes `yield* WorkflowEvent` and `yield* task(...)`
34 * instead of receiving callback parameters.
35 */
36export const makeWorkflowBridge =
37 (
38 WorkflowEntrypoint: abstract new (
39 ctx: unknown,
40 env: unknown,
41 ) => { run(event: any, step: any): Promise<unknown> },
42 {
43 entrypoint,
44 stack,
45 }: {
46 entrypoint: Effect.Effect<Record<string, any>>;
47 stack: { name: string; stage: string };
48 },
49 ) =>
50 (className: string) => {
51 // One isolate-lifetime layer build shared by every instantiation of this
52 // workflow class — `build` memoizes the built context.
53 const { build } = getWorkerExport<WorkflowExport>({
54 entrypoint,
55 stack,
56 exportName: className,
57 });
58
59 return class WorkflowBridge extends WorkflowEntrypoint {
60 readonly build: Promise<{
61 readonly context: Context.Context<never>;
62 readonly fn: WorkflowImpl<unknown, unknown, unknown>;
63 readonly telemetry: () => Layer.Layer<never, any, any> | undefined;
64 }>;
65
66 constructor(ctx: unknown, env: unknown) {
67 super(ctx, env);
68
69 this.build = build(() => {}).then(
70 ({ context, export: wf, telemetry }) =>
71 wf.make(env).pipe(
72 Effect.provideContext(context),
73 Effect.map((fn) => ({
74 context,
75 fn: fn as WorkflowImpl<unknown, unknown, unknown>,
76 telemetry,
77 })),
78 Effect.runPromise,
79 ),
80 );
81 }
82
83 async run(event: any, step: any): Promise<unknown> {
84 const { context, fn, telemetry } = await this.build;
85 // The run scope owns telemetry and resources outside tasks.
86 // Step attempts and rollback handlers use separate scopes.
87 const scope = Scope.makeUnsafe();
88 const exit = await Effect.runPromiseExit(
89 fn(event.payload).pipe(
90 Effect.provide(
91 Layer.mergeAll(
92 Layer.succeed(WorkflowEventService, wrapWorkflowEvent(event)),
93 Layer.succeed(
94 WorkflowStep,
95 wrapWorkflowStep(step, {
96 workflow: JSON.stringify([
97 stack.name,
98 stack.stage,
99 className,
100 event.workflowName ?? "",
101 ]),
102 instanceId: event.instanceId,
103 }),
104 ),
105 Layer.succeed(Scope.Scope, scope),
106 // The configured telemetry exporters, attached to the run's
107 // scope by `buildEventTelemetry` so buffered telemetry
108 // flushes when the scope closes at the end of the
109 // run-invocation.
110 Layer.effectContext(
111 buildEventTelemetry(context, scope, telemetry()),
112 ),
113 ).pipe(Layer.provideMerge(Layer.succeedContext(context))),
114 ),
115 ) as Effect.Effect<unknown, unknown>,
116 );
117 // Settle the run's resources with its real exit, unless a binding
118 // ejected the scope to outlive the invocation. The workflow runtime has
119 // no `waitUntil` to detach cleanup to, so close inline — a failing
120 // finalizer (e.g. a pg pool `end()` on a dropped connection) is logged
121 // and ignored so it can't mask the run's outcome.
122 if (!isScopeEjected(scope)) {
123 await Scope.close(scope, exit).pipe(
124 Effect.ignoreCause({
125 log: "Warn",
126 message: "Workflow run scope close failed",
127 }),
128 Effect.runPromise,
129 );
130 }
131 if (Exit.isSuccess(exit)) {
132 return exit.value;
133 }
134 throw Cause.squash(exit.cause);
135 }
136 };
137 };
138
139const wrapWorkflowEvent = (event: any): WorkflowEventService["Service"] => ({
140 payload: event.payload,
141 timestamp:
142 event.timestamp instanceof Date
143 ? event.timestamp
144 : new Date(event.timestamp),
145 instanceId: event.instanceId ?? "",
146 workflowName: event.workflowName ?? "",
147 schedule: event.schedule ?? undefined,
148});
149
150interface WorkflowIdentity {
151 readonly workflow: string;
152 readonly instanceId: string;
153}
154
155export const wrapWorkflowStep = (
156 step: any,
157 identity?: WorkflowIdentity,
158): WorkflowStep["Service"] => ({
159 do: <T, E>(
160 options: WorkflowTaskOptions<T, any, any, E>,
161 ): Effect.Effect<T, E> => {
162 const { name } = options;
163 // `task` provides application services; the bridge supplies attempt-local services.
164 const effect = options.effect as Effect.Effect<
165 T,
166 E,
167 WorkflowStepContext | Scope.Scope
168 >;
169 const config = definedStepConfig(options);
170 const rollbackEffect = options.rollback;
171 const rollback = rollbackEffect
172 ? {
173 // Native compensation may run after this step and the run scope have closed.
174 rollback: async (context: any) => {
175 const exit = await Effect.runPromiseExit(
176 Effect.scoped(
177 rollbackEffect({
178 error: context.error,
179 output: context.output,
180 }) as Effect.Effect<void, unknown, Scope.Scope>,
181 ),
182 );
183 if (Exit.isFailure(exit))
184 throw await callbackFailure(exit.cause, name, identity);
185 },
186 rollbackConfig: definedStepConfig(options.rollbackConfig),
187 }
188 : undefined;
189 return Effect.scoped(
190 Effect.gen(function* () {
191 // Join active callbacks on interruption, without waiting through native retry delays.
192 const runPromise = yield* FiberSet.makeRuntimePromise<
193 never,
194 Exit.Exit<T, E>
195 >();
196 let failure: { message: string; cause: Cause.Cause<E> } | undefined;
197 const callback = async (context: any) => {
198 const exit = await runPromise(
199 effect.pipe(
200 Effect.provideService(WorkflowStepContext, {
201 step: context.step,
202 attempt: context.attempt,
203 config: context.config,
204 }),
205 Effect.scoped,
206 Effect.exit,
207 ),
208 );
209 if (Exit.isSuccess(exit)) return exit.value;
210 const error = await callbackFailure(exit.cause, name, identity);
211 // Native RPC/storage discard custom fields. Correlate only this invocation's error.
212 if (!Cause.hasDies(exit.cause) && !Cause.hasInterrupts(exit.cause)) {
213 failure = { message: error.message, cause: exit.cause };
214 }
215 throw error;
216 };
217 return yield* Effect.tryPromise<T, unknown>({
218 try: () => {
219 if (config && rollback)
220 return step.do(name, config, callback, rollback);
221 if (config) return step.do(name, config, callback);
222 if (rollback) return step.do(name, callback, rollback);
223 return step.do(name, callback);
224 },
225 catch: (error) => error,
226 }).pipe(
227 Effect.catch((error) =>
228 failure &&
229 error instanceof Error &&
230 error.name === "Error" &&
231 error.message === failure.message
232 ? Effect.failCause(failure.cause)
233 : Effect.promise(() =>
234 decodeApplicationFailure<E>(error, name, identity),
235 ).pipe(
236 Effect.flatMap((cause) =>
237 cause ? Effect.failCause(cause) : Effect.die(error),
238 ),
239 ),
240 ),
241 );
242 }),
243 );
244 },
245 sleep: (name: string, duration: string | number): Effect.Effect<void> =>
246 Effect.promise(() => step.sleep(name, duration)),
247 sleepUntil: (name: string, timestamp: Date | number): Effect.Effect<void> =>
248 Effect.promise(() => step.sleepUntil(name, timestamp)),
249 waitForEvent: <T>(
250 name: string,
251 options: any,
252 ): Effect.Effect<WorkflowStepEvent<T>> =>
253 Effect.promise(
254 () => step.waitForEvent(name, options) as Promise<WorkflowStepEvent<T>>,
255 ),
256});
257
258const failurePrefix = "[alchemy-workflow-failure:v1]";
259const terminalPrefix = "[alchemy-workflow-terminal:v1]";
260const maxFailureLength = 16_384;
261
262const terminalFailure = async (message: string): Promise<Error> => {
263 const { NonRetryableError } = await import("cloudflare:workflows");
264 return new NonRetryableError(`${terminalPrefix}${message}`);
265};
266
267const callbackFailure = async <E>(
268 cause: Cause.Cause<E>,
269 step: string,
270 identity: WorkflowIdentity | undefined,
271): Promise<Error> => {
272 if (!Cause.hasDies(cause) && !Cause.hasInterrupts(cause)) {
273 try {
274 const budget = { nodes: 0, seen: new Set<object>() };
275 const errors = cause.reasons.map((reason) => {
276 if (reason._tag !== "Fail")
277 throw new TypeError("Expected an application failure");
278 return encodeFailureValue(reason.error, new Set<object>(), budget);
279 });
280 const error = Cause.squash(cause);
281 const summary =
282 error instanceof Error
283 ? error.message
284 : typeof error === "string"
285 ? error
286 : "Workflow application failure";
287 const body = `${Base64Url.encode(
288 JSON.stringify({
289 workflow: identity?.workflow ?? null,
290 instanceId: identity?.instanceId ?? null,
291 step,
292 errors,
293 }),
294 )}\n${summary}`;
295 const message = `${failurePrefix}${await failureDigest(body)}:${body}`;
296 if (new TextEncoder().encode(message).byteLength > maxFailureLength)
297 throw new TypeError("encoded failure exceeds 16 KiB");
298 return new Error(message);
299 } catch (error) {
300 throw await terminalFailure(
301 `Workflow application failure is not serializable: ${error instanceof Error ? error.message : "unsupported value"}`,
302 );
303 }
304 }
305 const error = Cause.squash(cause);
306 return terminalFailure(
307 error instanceof Error
308 ? error.message
309 : typeof error === "string"
310 ? error
311 : "Workflow callback defect or interruption",
312 );
313};
314
315const failureDigest = async (body: string): Promise<string> =>
316 Base64Url.encode(
317 new Uint8Array(
318 await crypto.subtle.digest("SHA-256", new TextEncoder().encode(body)),
319 ),
320 );
321
322const decodeApplicationFailure = async <E>(
323 error: unknown,
324 step: string,
325 identity: WorkflowIdentity | undefined,
326): Promise<Cause.Cause<E> | undefined> => {
327 if (
328 !identity?.workflow ||
329 !identity.instanceId ||
330 !(error instanceof Error) ||
331 error.name !== "Error" ||
332 !error.message.startsWith(failurePrefix)
333 )
334 return undefined;
335 try {
336 if (new TextEncoder().encode(error.message).byteLength > maxFailureLength)
337 throw new TypeError("encoded failure exceeds 16 KiB");
338 const encoded = error.message.slice(failurePrefix.length);
339 const separator = encoded.indexOf(":");
340 const digest = encoded.slice(0, separator);
341 const body = encoded.slice(separator + 1);
342 if (separator !== 43 || digest !== (await failureDigest(body)))
343 throw new TypeError("invalid failure checksum");
344 const end = body.indexOf("\n");
345 if (end < 1) throw new TypeError("invalid failure framing");
346 const json = Base64Url.decodeString(body.slice(0, end));
347 if (Result.isFailure(json)) throw new TypeError("invalid failure encoding");
348 const envelope: unknown = JSON.parse(json.success);
349 if (
350 !envelope ||
351 typeof envelope !== "object" ||
352 !("workflow" in envelope) ||
353 envelope.workflow !== identity.workflow ||
354 !("instanceId" in envelope) ||
355 envelope.instanceId !== identity.instanceId ||
356 !("step" in envelope) ||
357 envelope.step !== step ||
358 !("errors" in envelope) ||
359 !Array.isArray(envelope.errors) ||
360 envelope.errors.length === 0 ||
361 Object.keys(envelope).length !== 4
362 )
363 throw new TypeError("invalid failure envelope");
364 // Only the validated private transport restores E's data; prototypes and identity are invocation-local.
365 const errors = envelope.errors.map((value) =>
366 decodeFailureValue(value),
367 ) as E[];
368 return Cause.fromReasons(
369 errors.map((value) => Cause.makeFailReason(value)),
370 );
371 } catch (cause) {
372 throw new TypeError("Invalid persisted Workflow application failure", {
373 cause,
374 });
375 }
376};
377
378type FailureValue = null | boolean | string | number | FailureValue[];
379
380const encodeFailureValue = (
381 value: unknown,
382 ancestors = new Set<object>(),
383 budget = { nodes: 0, seen: new Set<object>() },
384): FailureValue => {
385 if (
386 ++budget.nodes > maxFailureLength ||
387 (typeof value === "string" && value.length > maxFailureLength)
388 )
389 throw new TypeError("encoded failure exceeds 16 KiB");
390 if (value === null || typeof value === "boolean" || typeof value === "string")
391 return value;
392 if (typeof value === "number")
393 return Number.isFinite(value) && !Object.is(value, -0)
394 ? value
395 : ["number", Object.is(value, -0) ? "-0" : String(value)];
396 if (value === undefined) return ["undefined"];
397 if (typeof value === "bigint") return ["bigint", String(value)];
398 if (typeof value !== "object")
399 throw new TypeError(`unsupported ${typeof value}`);
400 if (ancestors.has(value)) throw new TypeError("cyclic error data");
401 if (budget.seen.has(value))
402 throw new TypeError("shared references in error data");
403 budget.seen.add(value);
404 if (ancestors.size >= 64)
405 throw new TypeError("error data exceeds 64 nesting levels");
406 ancestors.add(value);
407 try {
408 const encode = (item: unknown) =>
409 encodeFailureValue(item, ancestors, budget);
410 const symbols = Object.getOwnPropertySymbols(value).filter(
411 (key) =>
412 !(
413 value instanceof Error &&
414 key === Symbol.for("effect/Data/Error/plainArgs")
415 ),
416 );
417 if (symbols.length) throw new TypeError("symbol-keyed error data");
418 if (
419 value instanceof Date ||
420 value instanceof Uint8Array ||
421 value instanceof ArrayBuffer ||
422 value instanceof Map ||
423 value instanceof Set
424 ) {
425 const own = Object.getOwnPropertyNames(value);
426 if (
427 own.some(
428 (key) => !(value instanceof Uint8Array && /^(0|[1-9]\d*)$/.test(key)),
429 )
430 )
431 throw new TypeError("custom properties on serialized built-in values");
432 if (value instanceof Date)
433 return ["date", encode(Date.prototype.getTime.call(value))];
434 if (value instanceof Uint8Array) return ["bytes", Base64.encode(value)];
435 if (value instanceof ArrayBuffer)
436 return ["buffer", Base64.encode(new Uint8Array(value))];
437 if (value instanceof Map)
438 return [
439 "map",
440 [...Map.prototype.entries.call(value)].map(([key, item]) => [
441 encode(key),
442 encode(item),
443 ]),
444 ];
445 return ["set", [...Set.prototype.values.call(value)].map(encode)];
446 }
447 if (Array.isArray(value)) {
448 const descriptors = Object.getOwnPropertyDescriptors(value);
449 const keys = Object.keys(descriptors);
450 if (
451 keys.length !== value.length + 1 ||
452 keys.some(
453 (key) =>
454 key !== "length" &&
455 (!/^(0|[1-9]\d*)$/.test(key) || Number(key) >= value.length),
456 )
457 )
458 throw new TypeError("sparse arrays or custom array properties");
459 for (const descriptor of Object.values(descriptors)) {
460 if (!("value" in descriptor))
461 throw new TypeError("accessor error data");
462 }
463 return [
464 "array",
465 Array.from({ length: value.length }, (_, i) =>
466 encode(descriptors[i].value),
467 ),
468 ];
469 }
470 const descriptors = Object.getOwnPropertyDescriptors(value);
471 let tag: string | undefined;
472 for (
473 let owner: object | null = value;
474 owner;
475 owner = Object.getPrototypeOf(owner)
476 ) {
477 const descriptor = Object.getOwnPropertyDescriptor(owner, "_tag");
478 if (!descriptor) continue;
479 if (!("value" in descriptor)) throw new TypeError("accessor error data");
480 if (typeof descriptor.value === "string") tag = descriptor.value;
481 break;
482 }
483 const prototype = Object.getPrototypeOf(value);
484 if (
485 !(value instanceof Error) &&
486 prototype !== null &&
487 prototype !== Object.prototype &&
488 tag === undefined
489 )
490 throw new TypeError("unsupported class instance");
491 for (const [key, descriptor] of Object.entries(descriptors)) {
492 if (
493 !("value" in descriptor) &&
494 !(value instanceof Error && key === "stack")
495 )
496 throw new TypeError("accessor error data");
497 }
498 const fields = new Map<string, unknown>();
499 if (tag !== undefined) fields.set("_tag", tag);
500 if (value instanceof Error) {
501 for (const key of ["name", "message", "stack"]) {
502 let owner: object | null = value;
503 while (owner) {
504 const descriptor = Object.getOwnPropertyDescriptor(owner, key);
505 if (descriptor) {
506 if (
507 !("value" in descriptor) &&
508 !(owner === value && key === "stack")
509 )
510 throw new TypeError("accessor error data");
511 break;
512 }
513 owner = Object.getPrototypeOf(owner);
514 }
515 }
516 for (const [key, item] of Object.entries(
517 encodeRpcError(value) as Record<string, unknown>,
518 ))
519 fields.set(key, item);
520 fields.set("name", value.name);
521 fields.set("message", value.message);
522 }
523 for (const [key, descriptor] of Object.entries(descriptors)) {
524 if (!("value" in descriptor)) {
525 if (value instanceof Error && key === "stack")
526 fields.set(key, value.stack);
527 else throw new TypeError("accessor error data");
528 } else fields.set(key, descriptor.value);
529 }
530 return [
531 value instanceof Error ? "error" : "object",
532 [...fields].map(([key, item]) => [key, encode(item)]),
533 ];
534 } finally {
535 ancestors.delete(value);
536 }
537};
538
539const decodeFailureValue = (value: unknown, depth = 0): unknown => {
540 if (
541 value === null ||
542 typeof value === "boolean" ||
543 typeof value === "string" ||
544 (typeof value === "number" && Number.isFinite(value))
545 )
546 return value;
547 if (!Array.isArray(value))
548 throw new TypeError("invalid serialized error value");
549 const decode = (item: unknown) => decodeFailureValue(item, depth + 1);
550 const [tag, data] = value;
551 if (tag === "undefined" && value.length === 1) return undefined;
552 if (value.length !== 2) throw new TypeError("invalid serialized error tuple");
553 if (
554 tag === "number" &&
555 typeof data === "string" &&
556 ["NaN", "Infinity", "-Infinity", "-0"].includes(data)
557 )
558 return Number(data);
559 if (
560 tag === "bigint" &&
561 typeof data === "string" &&
562 /^-?(0|[1-9]\d*)$/.test(data)
563 )
564 return BigInt(data);
565 if (depth >= 64) throw new TypeError("invalid serialized error value");
566 if (tag === "date") {
567 const timestamp = decode(data);
568 if (typeof timestamp === "number") return new Date(timestamp);
569 }
570 if ((tag === "bytes" || tag === "buffer") && typeof data === "string") {
571 const bytes = Base64.decode(data);
572 if (Result.isSuccess(bytes) && Base64.encode(bytes.success) === data)
573 return tag === "bytes" ? bytes.success : bytes.success.buffer;
574 }
575 if (Array.isArray(data)) {
576 if (tag === "array") return data.map(decode);
577 if (tag === "set") return new Set(data.map(decode));
578 if (tag === "map")
579 return new Map(
580 data.map((pair) => {
581 if (!Array.isArray(pair) || pair.length !== 2)
582 throw new TypeError("invalid serialized map");
583 return [decode(pair[0]), decode(pair[1])];
584 }),
585 );
586 if (tag === "object" || tag === "error") {
587 const result = tag === "error" ? new Error() : {};
588 const keys = new Set<string>();
589 for (const pair of data) {
590 if (
591 !Array.isArray(pair) ||
592 pair.length !== 2 ||
593 typeof pair[0] !== "string" ||
594 keys.has(pair[0])
595 )
596 throw new TypeError("invalid serialized error fields");
597 keys.add(pair[0]);
598 Object.defineProperty(result, pair[0], {
599 value: decode(pair[1]),
600 enumerable: true,
601 writable: true,
602 configurable: true,
603 });
604 }
605 return result;
606 }
607 }
608 throw new TypeError("invalid serialized error value");
609};
610
611// Own undefined properties overwrite the engine's defaults for steps and rollbacks.
612const definedStepConfig = (
613 options: WorkflowStepConfig | undefined,
614): WorkflowStepConfig | undefined => {
615 if (options === undefined) return undefined;
616 const config: WorkflowStepConfig = {};
617 if (options.retries !== undefined) config.retries = options.retries;
618 if (options.timeout !== undefined) config.timeout = options.timeout;
619 return Object.keys(config).length > 0 ? config : undefined;
620};