1import type * as runtime from "@cloudflare/workers-types";
2import * as workflows from "@distilled.cloud/cloudflare/workflows";
3import type { ConfigError } from "effect/Config";
4import * as Context from "effect/Context";
5import * as Effect from "effect/Effect";
6import { Scope } from "effect/Scope";
7import * as Stream from "effect/Stream";
8import { OwnedBySomeoneElse, Unowned } from "../../AdoptPolicy.ts";
9import { havePropsChanged, isResolved } from "../../Diff.ts";
10import type { Input } from "../../Input.ts";
11import * as ProviderLayer from "../../Local/ProviderLayer.ts";
12import type * as Output from "../../Output.ts";
13import { ALCHEMY_PHASE } from "../../Phase.ts";
14import type { PlatformServices } from "../../Platform.ts";
15import * as Provider from "../../Provider.ts";
16import { Resource } from "../../Resource.ts";
17import type { RuntimeContext } from "../../RuntimeContext.ts";
18import { effectClass, taggedFunction } from "../../Util/effect.ts";
19import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
20import { localAccountId } from "../LocalAccount.ts";
21import { generateLocalId } from "../LocalRuntime.ts";
22import {
23 Worker,
24 WorkerEnvironment,
25 type WorkerServices,
26} from "../Workers/Worker.ts";
27import { generateWorkflowName } from "./WorkflowName.ts";
28import {
29 type WorkflowEvent,
30 WorkflowStep,
31 WorkflowStepContext,
32} from "./WorkflowRuntime.ts";
33
34export {
35 WorkflowEvent,
36 WorkflowStep,
37 WorkflowStepContext,
38} from "./WorkflowRuntime.ts";
39
40type TypeId = "Cloudflare.Workflow";
41const TypeId = "Cloudflare.Workflow" as const;
42
43/**
44 * Cron trigger metadata on a Workflow instance created by a native
45 * `schedules` expression. Mirrors Cloudflare's `event.schedule`.
46 */
47export interface WorkflowCronSchedule {
48 /**
49 * The matching cron expression that created this instance.
50 */
51 cron: string;
52 /**
53 * The scheduled fire time, milliseconds since the Unix epoch.
54 */
55 scheduledTime: number;
56}
57
58export type WorkflowBackoff = "constant" | "linear" | "exponential";
59
60export interface WorkflowStepConfig {
61 retries?: {
62 limit: number;
63 delay: string | number;
64 backoff?: WorkflowBackoff;
65 };
66 timeout?: string | number;
67}
68
69export interface WorkflowStepContextData {
70 step: {
71 name: string;
72 count: number;
73 };
74 attempt: number;
75 config: WorkflowStepConfig;
76}
77
78export interface WorkflowRollbackContext<Output = unknown> {
79 error: Error;
80 output: Output | undefined;
81}
82
83export interface WorkflowRollbackOptions<Output = unknown, R = never> {
84 rollback: (
85 context: WorkflowRollbackContext<Output>,
86 ) => Effect.Effect<void, unknown, R>;
87 rollbackConfig?: WorkflowStepConfig;
88}
89
90/**
91 * Optional configuration for a `task` step: retry policy, timeout, and a
92 * rollback handler with its own retry config.
93 */
94export interface WorkflowTaskConfig<
95 Output = unknown,
96 RollbackReq = never,
97> extends WorkflowStepConfig {
98 rollback?: (
99 context: WorkflowRollbackContext<Output>,
100 ) => Effect.Effect<void, unknown, RollbackReq>;
101 rollbackConfig?: WorkflowStepConfig;
102}
103
104/**
105 * Internal step descriptor passed from `task` to the bridge. Bundles the step
106 * name and Effect together with the `WorkflowTaskConfig` fields.
107 */
108export interface WorkflowTaskOptions<
109 Output = unknown,
110 R = never,
111 RollbackReq = never,
112 E = never,
113> extends WorkflowTaskConfig<Output, RollbackReq> {
114 name: string;
115 effect: Effect.Effect<Output, E, R>;
116}
117
118export interface WorkflowWaitForEventOptions {
119 type: string;
120 timeout?: string | number;
121}
122
123/**
124 * The event delivered to a `waitForEvent` step. Mirrors the native
125 * `WorkflowStepEvent` shape from `cloudflare:workers` 1:1.
126 */
127export interface WorkflowStepEvent<Payload = unknown> {
128 payload: Payload;
129 timestamp: Date;
130 type: string;
131}
132
133type ExcludeWorkflowStepContext<R> = R extends {
134 readonly key: "Cloudflare.WorkflowStepContext";
135}
136 ? never
137 : R;
138
139// ---------------------------------------------------------------------------
140// User-facing step primitives
141// ---------------------------------------------------------------------------
142
143/**
144 * Execute a named, durable workflow step. The effect is run inside the
145 * Cloudflare step transaction so its result is automatically persisted
146 * and replayed on retries.
147 *
148 * Any services the inner effect requires (e.g. `WorkerEnvironment` from a
149 * binding like `kv.put` / `kv.get`) are threaded through automatically by
150 * capturing the surrounding workflow body's context and providing it to
151 * the inner effect before it runs inside `step.do`.
152 *
153 * The step name comes first, followed by the Effect. Retry config, timeout,
154 * and a rollback handler can be passed in the optional third `options` arg.
155 * Each attempt and rollback handler has a fresh Scope; its resources close
156 * before that callback completes. Interrupting a task waits for the active
157 * attempt's cleanup without waiting for Cloudflare's native retry delays.
158 *
159 * `Effect.fail` uses the native retry policy. On exhaustion, application failure
160 * data is returned in the error channel, including on cached rejection replay.
161 * The active invocation retains its original Cause; replay restores serialized
162 * tags and fields, not custom prototypes, methods, or object identity.
163 * Failure data supports primitives, dense arrays, records, errors, Date, bigint,
164 * Uint8Array, ArrayBuffer, Map, and Set, within a 16 KiB encoded / 64-level limit.
165 * Functions, symbols, cycles, shared object references, accessors, and unsupported
166 * class instances fail explicitly as terminal serialization defects. Error data
167 * must form a tree, including objects used as Map keys or Set members.
168 * `Effect.die` and `Effect.orDie`
169 * stop retries. Native timeout, validation, pause, and abort errors stay defects.
170 */
171export function task<T, R = never, RollbackReq = never, E = never>(
172 name: string,
173 effect: Effect.Effect<T, E, R>,
174 options?: WorkflowTaskConfig<T, RollbackReq>,
175): Effect.Effect<
176 T,
177 E,
178 WorkflowStep | ExcludeWorkflowStepContext<Exclude<R | RollbackReq, Scope>>
179> {
180 return Effect.gen(function* () {
181 const step = yield* WorkflowStep;
182 const context = (yield* Effect.context<
183 ExcludeWorkflowStepContext<Exclude<R | RollbackReq, Scope>>
184 >()).pipe(Context.omit(Scope, WorkflowStepContext));
185 const rollbackEffect = options?.rollback;
186 return yield* step.do({
187 ...options,
188 name,
189 effect: effect.pipe(Effect.provide(context)),
190 rollback: rollbackEffect
191 ? (rollbackContext: WorkflowRollbackContext<T>) =>
192 rollbackEffect(rollbackContext).pipe(Effect.provide(context))
193 : undefined,
194 } as WorkflowTaskOptions<T, any, any, E>);
195 });
196}
197
198/**
199 * Pause the workflow for the given duration.
200 */
201export const sleep = (
202 name: string,
203 duration: string | number,
204): Effect.Effect<void, never, WorkflowStep> =>
205 Effect.gen(function* () {
206 const step = yield* WorkflowStep;
207 yield* step.sleep(name, duration);
208 }).pipe(Effect.orDie);
209
210/**
211 * Pause the workflow until the given timestamp.
212 */
213export const sleepUntil = (
214 name: string,
215 timestamp: Date | number,
216): Effect.Effect<void, never, WorkflowStep> =>
217 Effect.gen(function* () {
218 const step = yield* WorkflowStep;
219 yield* step.sleepUntil(name, timestamp);
220 }).pipe(Effect.orDie);
221
222/**
223 * Pause the workflow until an external event is delivered with
224 * `WorkflowInstance.sendEvent`. Resolves with the full
225 * {@link WorkflowStepEvent} (`{ payload, timestamp, type }`), exactly like
226 * the native `step.waitForEvent`.
227 */
228export const waitForEvent = <T = unknown>(
229 name: string,
230 options: WorkflowWaitForEventOptions,
231): Effect.Effect<WorkflowStepEvent<T>, never, WorkflowStep> =>
232 Effect.gen(function* () {
233 const step = yield* WorkflowStep;
234 return yield* step.waitForEvent<T>(name, options);
235 }).pipe(Effect.orDie);
236
237/**
238 * The services available inside a workflow run body.
239 *
240 * `WorkerEnvironment` is provided to the body at runtime by the workflow
241 * export wrapper (see `make(env)` below), so users can access env bindings
242 * from inside workflow steps via `yield* WorkerEnvironment` — the type must
243 * reflect that or `yield* WorkerEnvironment` fails to type-check inside a
244 * body even though it succeeds at runtime.
245 *
246 * A fresh `Scope` is provided per run-invocation by `WorkflowBridge.run` and
247 * threaded into every `task` via the surrounding body context, so `@binding`
248 * helpers that acquire per-run resources against the ambient scope (e.g.
249 * `Drizzle.Postgres`) resolve them inside workflow steps just as they do in
250 * a Worker `fetch`/`queue` handler.
251 */
252export type WorkflowRunServices =
253 | WorkflowEvent
254 | WorkflowStep
255 | WorkerServices
256 | Scope;
257
258export type WorkflowServices =
259 | WorkflowRunServices
260 | PlatformServices
261 | RuntimeContext;
262
263/**
264 * Metadata stored in the worker export map to distinguish workflow exports
265 * from durable object exports at bundle-generation time.
266 */
267export interface WorkflowExport {
268 readonly kind: "workflow";
269 readonly make: (
270 env: unknown,
271 ) => Effect.Effect<WorkflowImpl<any, any, unknown>>;
272}
273
274/**
275 * A workflow implementation is a function from a typed `Input` payload to
276 * an Effect that produces the workflow's `Result`. The Effect requires
277 * `WorkflowRunServices` (event + step + env) to execute.
278 */
279export type WorkflowImpl<Input = unknown, Result = unknown, E = never> = (
280 input: Input,
281) => Effect.Effect<Result, E, WorkflowServices>;
282
283export const isWorkflowExport = (value: unknown): value is WorkflowExport =>
284 typeof value === "object" &&
285 value !== null &&
286 "kind" in value &&
287 (value as any).kind === "workflow";
288
289/**
290 * Limits applied to the workflow on create or update.
291 */
292export interface WorkflowLimits {
293 /**
294 * Maximum number of steps a single workflow instance may execute.
295 */
296 steps?: number;
297}
298
299/**
300 * Props for the reference (async) form of {@link Workflow}. Used when binding
301 * a Workflow class to a plain async Worker (one without an Effect runtime) via
302 * the Worker's `env`. Mirrors `DurableObjectProps`.
303 */
304export interface WorkflowRefProps {
305 /**
306 * Account-global Workflow name. Adopting an existing Workflow requires
307 * `--adopt`; unused names do not. If omitted, Alchemy preserves the deployed
308 * name or derives a stage-scoped default from the hosting Worker and class.
309 * Changing this name replaces the Workflow. Fixed names must be unique to
310 * each independently managed stack and stage.
311 */
312 workflowName?: string;
313 /**
314 * Name of the exported `WorkflowEntrypoint` class.
315 *
316 * @default name
317 */
318 className?: string;
319 /**
320 * Worker script that hosts the Workflow class. Omit this when the workflow
321 * is hosted by the Worker that declares the binding.
322 */
323 scriptName?: Input<string>;
324 /**
325 * Limits applied to the workflow. Only applies when the workflow is hosted by
326 * the Worker that declares the binding; ignored when `scriptName` is set.
327 */
328 limits?: WorkflowLimits;
329 /**
330 * Cron expressions that create a new Workflow instance on each match.
331 * Wrangler-compatible: `schedules: ["0 * * * *"]`.
332 *
333 * Native Workflow schedules replace a Worker Cron Trigger whose
334 * `scheduled` handler called `workflow.create()`. Pass an empty
335 * array to remove all schedules.
336 *
337 * Only applies when the workflow is hosted by the Worker that declares
338 * the binding; ignored when `scriptName` is set.
339 *
340 * Account-wide limit: 100 cron expressions.
341 */
342 schedules?: string[];
343}
344
345/**
346 * Props for the Effect-native form of {@link Workflow}
347 * (`Workflow(name, props, impl)`). Used when the workflow's implementation is
348 * defined inline by the hosting Worker.
349 */
350export interface WorkflowProps {
351 /**
352 * Account-global Workflow name. Adopting an existing Workflow requires
353 * `--adopt`; unused names do not. If omitted, Alchemy preserves the deployed
354 * name or derives a stage-scoped default from the hosting Worker and class.
355 * Changing this name replaces the Workflow. Fixed names must be unique to
356 * each independently managed stack and stage.
357 */
358 workflowName?: string;
359 /**
360 * Limits applied to the workflow.
361 */
362 limits?: WorkflowLimits;
363 /**
364 * Cron expressions that create a new Workflow instance on each match.
365 * Wrangler-compatible: `schedules: ["0 * * * *"]`.
366 *
367 * Native Workflow schedules replace a Worker Cron Trigger whose
368 * `scheduled` handler called `workflow.create()`. Pass an empty
369 * array to remove all schedules.
370 *
371 * Account-wide limit: 100 cron expressions.
372 */
373 schedules?: string[];
374}
375
376/**
377 * A lightweight reference to a Workflow, produced by the props-only form of
378 * {@link Workflow} (`Workflow(name, { className })`). Carries just enough
379 * metadata to emit the `workflow` binding for an async Worker and to drive
380 * the `putWorkflow` lifecycle. Mirrors `DurableObjectLike`.
381 */
382export interface WorkflowLike<Params = unknown> {
383 kind: TypeId;
384 name: string;
385 /** Account-global Workflow name, when explicitly configured. */
386 workflowName?: string;
387 /** @internal phantom */
388 className?: string;
389 /** @internal phantom */
390 scriptName?: Input<string>;
391 /** @internal phantom */
392 limits?: WorkflowLimits;
393 /** @internal phantom */
394 schedules?: string[];
395 /** @internal phantom */
396 Params?: Params;
397}
398
399/**
400 * A Workflow bound on an external/async Worker's `env`, exposed on
401 * `worker.env.<name>` at declaration time, not on persisted Worker references
402 * or Effect-native Workers. Carries the binding's identity as `Output`s
403 * of the current deploy pass, so a sibling resource in the same stack — a
404 * Queue subscription to the Workflow's lifecycle events, for instance — can
405 * consume the Workflow's physical name on its very first deployment without
406 * reading persisted state or re-deriving the name.
407 *
408 * For a locally-hosted Workflow, `workflowName` and `scriptName` resolve from
409 * the {@link WorkflowResource} the binding registers, so a consumer deploys
410 * after `putWorkflow` has run. A cross-script reference (`scriptName` set)
411 * registers no resource; both outputs derive from the declared host script.
412 */
413export interface WorkflowBinding<Params = unknown> {
414 kind: TypeId;
415 /** Logical name of the Workflow, as passed to `Workflow(name, …)`. */
416 name: string;
417 /** Name of the exported `WorkflowEntrypoint` class. */
418 className: string;
419 /** Account-global physical Workflow name, resolved in the current deploy. */
420 workflowName: Output.Output<string>;
421 /** Script name of the Worker hosting the Workflow class. */
422 scriptName: Output.Output<string>;
423 /** @internal phantom */
424 Params?: Params;
425}
426
427/**
428 * Type guard for the reference (async) form of a Workflow.
429 */
430export const isWorkflowLike = (value: unknown): value is WorkflowLike =>
431 typeof value === "object" &&
432 value !== null &&
433 (value as { kind?: unknown }).kind === TypeId;
434
435/**
436 * Type guard for workflow binding metadata in the Worker binding contract.
437 */
438export const isWorkflowBinding = (binding: {
439 type: string;
440}): binding is {
441 type: "workflow";
442 name: string;
443 workflowName: string;
444 className: string;
445 scriptName?: string;
446} => binding.type === "workflow";
447
448export type WorkflowBatchDeleteResult = runtime.WorkflowBatchDeleteResult;
449export type WorkflowSubscriptionEvent = runtime.WorkflowInstanceEvent;
450export type WorkflowInstanceSubscribeOptions =
451 runtime.WorkflowInstanceSubscribeOptions;
452
453/**
454 * Handle returned to the caller at deploy/bind time. Allows starting
455 * workflow instances and checking their status from the Api layer.
456 */
457export interface WorkflowHandle<Input = unknown, Result = unknown> {
458 Type: TypeId;
459 name: string;
460 /**
461 * Start a workflow instance. Pass payload through `params`; omit `id` to let
462 * Cloudflare generate an instance ID.
463 */
464 create(
465 options?: WorkflowInstanceCreateOptions<Input>,
466 ): Effect.Effect<WorkflowInstance<Result>>;
467 createBatch(
468 batch: WorkflowInstanceCreateOptions<Input>[],
469 ): Effect.Effect<WorkflowInstance<Result>[]>;
470 get(instanceId: string): Effect.Effect<WorkflowInstance<Result>>;
471 /** Delete up to 100 instances and their stored state, returning per-instance results. */
472 deleteBatch(instanceIds: string[]): Effect.Effect<WorkflowBatchDeleteResult>;
473}
474
475/** Options for starting a workflow instance. */
476export interface WorkflowInstanceCreateOptions<Input = unknown> {
477 id?: string;
478 params?: Input;
479 retention?: WorkflowInstanceRetention;
480}
481
482export interface WorkflowInstanceRetention {
483 successRetention?: string | number;
484 errorRetention?: string | number;
485}
486
487/** Handle for a single Cloudflare workflow instance. */
488export interface WorkflowInstance<Result = unknown> {
489 id: string;
490 status(): Effect.Effect<WorkflowInstanceStatus<Result>>;
491 pause(): Effect.Effect<void>;
492 resume(): Effect.Effect<void>;
493 restart(options?: WorkflowInstanceRestartOptions): Effect.Effect<void>;
494 terminate(): Effect.Effect<void>;
495 /** Stop execution and delete this instance and its stored state. */
496 delete(): Effect.Effect<void>;
497 /** Stream historical and new events; release the subscription when consumption ends. */
498 subscribe(
499 options?: WorkflowInstanceSubscribeOptions,
500 ): Stream.Stream<WorkflowSubscriptionEvent>;
501 sendEvent<Event = unknown>(
502 event: WorkflowInstanceEvent<Event>,
503 ): Effect.Effect<void>;
504}
505
506export interface WorkflowInstanceRestartOptions {
507 from?: {
508 name: string;
509 count?: number;
510 type?: "do" | "sleep" | "waitForEvent";
511 };
512}
513
514export interface WorkflowInstanceEvent<Payload = unknown> {
515 type: string;
516 payload?: Payload;
517}
518
519export interface WorkflowInstanceStatus<Result = unknown> {
520 status:
521 | "queued"
522 | "running"
523 | "paused"
524 | "errored"
525 | "terminated"
526 | "complete"
527 | "waiting"
528 | "waitingForPause"
529 | "unknown"
530 | (string & {});
531 output?: Result;
532 error?: { name: string; message: string } | null;
533 rollback?: {
534 outcome: "complete" | "failed";
535 error: { name: string; message: string } | null;
536 } | null;
537}
538
539export interface WorkflowClass extends Effect.Effect<
540 WorkflowHandle,
541 never,
542 WorkflowHandle
543> {
544 /** Reference a deployed Workflow by logical ID, optionally in another stack or stage. */
545 ref: typeof WorkflowResource.ref;
546 <_Self>(): {
547 <Input = unknown, Result = unknown, InitReq = never, E = never>(
548 name: string,
549 impl: Effect.Effect<WorkflowImpl<Input, Result, E>, ConfigError, InitReq>,
550 ): Effect.Effect<
551 WorkflowHandle<Input, Result>,
552 never,
553 Worker | Exclude<InitReq, WorkflowServices>
554 > & {
555 new (_: never): WorkflowImpl<Input, Result, E>;
556 };
557 <Input = unknown, Result = unknown, InitReq = never, E = never>(
558 name: string,
559 props: WorkflowProps,
560 impl: Effect.Effect<WorkflowImpl<Input, Result, E>, ConfigError, InitReq>,
561 ): Effect.Effect<
562 WorkflowHandle<Input, Result>,
563 never,
564 Worker | Exclude<InitReq, WorkflowServices>
565 > & {
566 new (_: never): WorkflowImpl<Input, Result, E>;
567 };
568 };
569 <Params = unknown>(
570 name: string,
571 props?: WorkflowRefProps,
572 ): WorkflowLike<Params>;
573 <Input = unknown, Result = unknown, InitReq = never, E = never>(
574 name: string,
575 impl: Effect.Effect<WorkflowImpl<Input, Result, E>, ConfigError, InitReq>,
576 ): Effect.Effect<
577 WorkflowHandle<Input, Result>,
578 never,
579 Worker | Exclude<InitReq, WorkflowServices>
580 >;
581 <Input = unknown, Result = unknown, InitReq = never, E = never>(
582 name: string,
583 props: WorkflowProps,
584 impl: Effect.Effect<WorkflowImpl<Input, Result, E>, ConfigError, InitReq>,
585 ): Effect.Effect<
586 WorkflowHandle<Input, Result>,
587 never,
588 Worker | Exclude<InitReq, WorkflowServices>
589 >;
590}
591
592export class WorkflowScope extends Context.Service<
593 WorkflowScope,
594 WorkflowHandle
595>()("Cloudflare.Workflow") {}
596
597/**
598 * A Cloudflare Workflow that orchestrates durable, multi-step tasks with
599 * automatic retries and at-least-once delivery.
600 *
601 * A Workflow follows the same two-phase pattern as Workers and Durable
602 * Objects. The outer `Effect.gen` resolves shared dependencies. The inner
603 * `Effect.fn` is the workflow body — a function from a typed `input`
604 * payload to an Effect that runs steps using `task`, `sleep`, and
605 * `sleepUntil`. `task` takes the step name and Effect, plus an optional
606 * config object for retries, timeout, and a rollback handler.
607 *
608 * ```typescript
609 * Effect.gen(function* () {
610 * // Phase 1: resolve dependencies
611 * const notifier = yield* NotificationService;
612 *
613 * return Effect.fn(function* (input: { orderId: string }) {
614 * // Phase 2: workflow body (durable steps)
615 * const result = yield* Cloudflare.Workflows.task("process", doWork(input.orderId));
616 * yield* Cloudflare.Workflows.sleep("cooldown", "10 seconds");
617 * return result;
618 * });
619 * })
620 * ```
621 *
622 *
623 * ### Defining a Workflow
624 * **Example:** Minimal workflow
625 * ```typescript
626 * export default class MyWorkflow extends Cloudflare.Workflow<MyWorkflow>()(
627 * "MyWorkflow",
628 * Effect.gen(function* () {
629 * return Effect.fn(function* (input: { name: string }) {
630 * return { received: input.name };
631 * });
632 * }),
633 * ) {}
634 * ```
635 *
636 * **Example:** Preserving an existing Workflow name
637 *
638 * An existing Workflow has no ownership marker, so the first deployment must
639 * opt in with `--adopt`. Keep fixed names unique per independently managed
640 * stack and stage; unlike the default, they are not stage-scoped by Alchemy.
641 *
642 * ```typescript
643 * export default class MyWorkflow extends Cloudflare.Workflow<MyWorkflow>()(
644 * "MyWorkflow",
645 * { workflowName: "my-existing-workflow" },
646 * Effect.gen(function* () {
647 * return Effect.fn(function* (input: { name: string }) {
648 * return { received: input.name };
649 * });
650 * }),
651 * ) {}
652 * ```
653 *
654 * **Example:** Setting a step limit
655 * ```typescript
656 * export default class MyWorkflow extends Cloudflare.Workflow<MyWorkflow>()(
657 * "MyWorkflow",
658 * { limits: { steps: 25000 } },
659 * Effect.gen(function* () {
660 * return Effect.fn(function* (input: { name: string }) {
661 * return { received: input.name };
662 * });
663 * }),
664 * ) {}
665 * ```
666 *
667 * ### Scheduling instances
668 * Each matching cron expression creates a new instance automatically —
669 * no Worker Cron Trigger or `scheduled` handler required.
670 * **Example:** Native cron schedules
671 * ```typescript
672 * export default class HourlyWorkflow extends Cloudflare.Workflow<HourlyWorkflow>()(
673 * "HourlyWorkflow",
674 * { schedules: ["0 * * * *"] },
675 * Effect.gen(function* () {
676 * return Effect.fn(function* () {
677 * const event = yield* Cloudflare.Workflows.WorkflowEvent;
678 * if (event.schedule) {
679 * return { cron: event.schedule.cron };
680 * }
681 * return {};
682 * });
683 * }),
684 * ) {}
685 * ```
686 *
687 * ### Step Primitives
688 * **Example:** Running a named task
689 * ```typescript
690 * const result = yield* Cloudflare.Workflows.task(
691 * "process-order",
692 * Effect.succeed({ orderId: "abc", total: 42 }),
693 * );
694 * ```
695 *
696 * **Example:** Configuring retries and reading step context
697 * ```typescript
698 * const result = yield* Cloudflare.Workflows.task(
699 * "call-api",
700 * Effect.gen(function* () {
701 * const context = yield* Cloudflare.Workflows.WorkflowStepContext;
702 * return { attempt: context.attempt };
703 * }),
704 * { retries: { limit: 3, delay: "5 seconds", backoff: "linear" } },
705 * );
706 * ```
707 *
708 * **Example:** Registering rollback
709 * ```typescript
710 * yield* Cloudflare.Workflows.task("reserve-inventory", reserveInventory, {
711 * rollback: ({ output }) =>
712 * output ? releaseInventory(output.reservationId) : Effect.void,
713 * rollbackConfig: { retries: { limit: 3, delay: "10 seconds" } },
714 * });
715 * ```
716 *
717 * **Example:** Sleeping between steps
718 * ```typescript
719 * yield* Cloudflare.Workflows.sleep("cooldown", "30 seconds");
720 * ```
721 *
722 * **Example:** Waiting for an external event
723 * ```typescript
724 * const event = yield* Cloudflare.Workflows.waitForEvent<{ approved: boolean }>(
725 * "approval",
726 * { type: "approval", timeout: "1 day" },
727 * );
728 * // Same shape as the native step.waitForEvent result:
729 * event.payload.approved;
730 * ```
731 *
732 * **Example:** Accessing env bindings inside a task
733 * Bind a resource (e.g. `Namespace`, `Bucket`) in the workflow's
734 * outer init phase to get a typed Effect-native client, then use it
735 * directly inside `task`. `task` threads the binding's service
736 * requirement (`WorkerEnvironment`) through automatically so the inner
737 * Effect needs no extra plumbing.
738 *
739 * ```typescript
740 * Effect.gen(function* () {
741 * const kv = yield* Cloudflare.KV.ReadWriteNamespace(KV);
742 *
743 * return Effect.fn(function* (input: { roomId: string; message: string }) {
744 * const { roomId, message } = input;
745 *
746 * const stored = yield* Cloudflare.Workflows.task(
747 * "kv-roundtrip",
748 * Effect.gen(function* () {
749 * const key = `workflow:${roomId}`;
750 * yield* kv.put(key, message);
751 * return yield* kv.get(key);
752 * }).pipe(Effect.orDie),
753 * );
754 *
755 * return stored;
756 * });
757 * });
758 * ```
759 *
760 * ### Starting and Monitoring Instances
761 * `create` mirrors Cloudflare's native Workflow API: pass workflow input in
762 * `params`, pass `id` only when you need a deterministic instance ID, and omit
763 * `id` to let Cloudflare generate one.
764 *
765 * **Example:** Creating an instance from a Worker
766 * ```typescript
767 * const workflow = yield* MyWorkflow;
768 * const instance = yield* workflow.create({ params: { orderId: "abc" } });
769 * ```
770 *
771 * **Example:** Creating an instance with id and retention
772 * ```typescript
773 * const instance = yield* workflow.create({
774 * id: "order-abc",
775 * params: { orderId: "abc" },
776 * retention: { successRetention: "1 day", errorRetention: "7 days" },
777 * });
778 * ```
779 *
780 * **Example:** Creating a batch
781 * ```typescript
782 * const instances = yield* workflow.createBatch([
783 * { id: "order-a", params: { orderId: "a" } },
784 * { id: "order-b", params: { orderId: "b" } },
785 * ]);
786 * ```
787 *
788 * **Example:** Checking instance status
789 * ```typescript
790 * const workflow = yield* MyWorkflow;
791 * const handle = yield* workflow.get(instanceId);
792 * const status = yield* handle.status();
793 * ```
794 *
795 * **Example:** Sending events and restarting instances
796 * ```typescript
797 * const instance = yield* workflow.get(instanceId);
798 * yield* instance.sendEvent({ type: "approval", payload: { approved: true } });
799 * yield* instance.restart({ from: { name: "approval", type: "waitForEvent" } });
800 * ```
801 *
802 * ### Triggering from a Worker
803 * Wire the workflow into HTTP routes so callers can fire instances
804 * and poll for completion.
805 *
806 * **Example:** Workflow start + status routes
807 * ```typescript
808 * // src/worker.ts
809 * const notifier = yield* MyWorkflow;
810 *
811 * return {
812 * fetch: Effect.gen(function* () {
813 * const request = yield* HttpServerRequest;
814 *
815 * if (request.url.startsWith("/workflow/start/")) {
816 * const id = request.url.split("/").pop()!;
817 * const instance = yield* notifier.create({ params: { orderId: id } });
818 * return HttpServerResponse.json({ instanceId: instance.id });
819 * }
820 *
821 * if (request.url.startsWith("/workflow/status/")) {
822 * const id = request.url.split("/").pop()!;
823 * const instance = yield* notifier.get(id);
824 * return HttpServerResponse.json(yield* instance.status());
825 * }
826 *
827 * return HttpServerResponse.text("Not Found", { status: 404 });
828 * }),
829 * };
830 * ```
831 *
832 * ### Binding in an Async Worker
833 * When using an Async Worker (plain `async fetch` handler, no Effect
834 * runtime), declare Workflows in the `env` prop of the Worker resource.
835 * Pass a `Workflow` reference with a `className` matching the exported
836 * `WorkflowEntrypoint` subclass in your worker source file. If `className`
837 * is omitted, it defaults to the binding name. Use `Cloudflare.InferEnv`
838 * to get a fully typed `env` object that includes the workflow binding.
839 *
840 * **Example:** Declaring a Workflow binding in the stack
841 * ```typescript
842 * // alchemy.run.ts
843 * export type WorkerEnv = Cloudflare.InferEnv<typeof Worker>;
844 *
845 * export const Worker = Cloudflare.Worker("Worker", {
846 * main: "./src/worker.ts",
847 * env: {
848 * MY_WORKFLOW: Cloudflare.Workflow<{ value: string }>("MyWorkflow", {
849 * className: "MyWorkflow",
850 * workflowName: "my-existing-workflow",
851 * }),
852 * },
853 * });
854 * ```
855 *
856 * **Example:** Native cron schedules on an async Workflow binding
857 * ```typescript
858 * export const Worker = Cloudflare.Worker("Worker", {
859 * main: "./src/worker.ts",
860 * env: {
861 * HOURLY: Cloudflare.Workflow("HourlyWorkflow", {
862 * className: "HourlyWorkflow",
863 * schedules: ["0 * * * *"],
864 * }),
865 * },
866 * });
867 * ```
868 *
869 * **Example:** Using the Workflow from a plain async handler
870 * ```typescript
871 * // src/worker.ts
872 * import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloudflare:workers";
873 * import type { WorkerEnv } from "../alchemy.run.ts";
874 *
875 * export class MyWorkflow extends WorkflowEntrypoint<WorkerEnv, { value: string }> {
876 * async run(event: Readonly<WorkflowEvent<{ value: string }>>, step: WorkflowStep) {
877 * return await step.do("greet", async () => `Hello, ${event.payload.value}!`);
878 * }
879 * }
880 *
881 * export default {
882 * async fetch(request: Request, env: WorkerEnv) {
883 * const instance = await env.MY_WORKFLOW.create({ params: { value: "world" } });
884 * return Response.json({ instanceId: instance.id });
885 * },
886 * };
887 * ```
888 *
889 * ### Consuming the Workflow's Physical Name
890 * An external/async Worker declaration exposes each Workflow binding on `worker.env` with
891 * the Workflow's account-global `workflowName` as an `Output` of the
892 * current deploy. Pass the binding directly as a Queue subscription's
893 * `source`; it deploys after the Workflow is registered, including on the
894 * first deployment. Explicit source descriptors remain supported for
895 * Workflows referenced by physical name.
896 *
897 * **Example:** Subscribing a Queue to the Workflow's events
898 * ```typescript
899 * const worker = yield* Cloudflare.Worker("Worker", {
900 * main: "./src/worker.ts",
901 * env: {
902 * FILE_URL_INGESTION: Cloudflare.Workflow("FileUrlIngestion", {
903 * className: "FileUrlIngestionWorkflow",
904 * }),
905 * },
906 * });
907 *
908 * yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
909 * source: worker.env.FILE_URL_INGESTION,
910 * events: ["instance.completed", "instance.errored"],
911 * queueId: queue.queueId,
912 * });
913 * ```
914 *
915 * ### Referencing a Deployed Workflow
916 * `Workflow.ref` reads the same persisted resource as `WorkflowResource.ref`.
917 * Use its logical ID (including any namespace), not its physical name or
918 * Worker env key. Omitting `stack` and `stage` uses the current stack/stage.
919 * Deploy the host first; a reference does not register or own the Workflow
920 * and returns resource attributes, not a runtime `WorkflowHandle`.
921 *
922 * **Example:** Subscribe to a Workflow in another stack
923 * ```typescript
924 * yield* Cloudflare.Queues.Subscription("WorkflowEvents", {
925 * source: yield* Cloudflare.Workflow.ref("Ingestion", {
926 * stack: "workflow-host",
927 * stage: "production",
928 * }),
929 * events: ["instance.completed", "instance.errored"],
930 * queueId: queue.queueId,
931 * });
932 * ```
933 *
934 * ### Cross-Script Binding in an Async Worker
935 * Async Workers can also bind to a Workflow hosted by another Worker
936 * script. The host Worker declares and exports the `WorkflowEntrypoint`
937 * class. The consumer Worker declares a `Workflow` with `scriptName` set
938 * to the host Worker's script name. Cross-script references are bindings
939 * only — Alchemy does not drive `putWorkflow` for the foreign class, so
940 * deploy the host first.
941 *
942 * **Example:** Consumer Worker binds to the host script
943 * ```typescript
944 * const consumer = yield* Cloudflare.Worker("Consumer", {
945 * main: "./src/consumer.ts",
946 * env: {
947 * MY_WORKFLOW: Cloudflare.Workflow("MyWorkflow", {
948 * className: "MyWorkflow",
949 * scriptName: host.workerName,
950 * // Repeat the host's explicit or preserved physical name when it
951 * // does not use Alchemy's current generated default.
952 * workflowName: "my-existing-workflow",
953 * }),
954 * },
955 * });
956 * ```
957 *
958 * ### Testing Workflows
959 * Workflows run asynchronously, so tests start an instance and poll until it
960 * reaches a terminal status. Keep polling bounded with `Effect.repeat`.
961 *
962 * **Example:** Polling for workflow completion
963 * ```typescript
964 * test(
965 * "workflow completes",
966 * Effect.gen(function* () {
967 * const { url } = yield* stack;
968 *
969 * const start = yield* HttpClient.post(`${url}/workflow/start/x`);
970 * const { instanceId } = (yield* start.json) as { instanceId: string };
971 *
972 * const status = yield* HttpClient.get(
973 * `${url}/workflow/status/${instanceId}`,
974 * ).pipe(
975 * Effect.flatMap((res) => res.json),
976 * Effect.map((json) => json as { status: string }),
977 * Effect.repeat({
978 * schedule: Schedule.spaced("2 seconds"),
979 * until: (status) =>
980 * status.status === "complete" || status.status === "errored",
981 * times: 30,
982 * }),
983 * );
984 *
985 * expect(status.status).toBe("complete");
986 * }),
987 * { timeout: 120_000 },
988 * );
989 * ```
990 *
991 * ### Observing and Deleting Instances
992 * **Example:** Read an event and remove stored state
993 * ```typescript
994 * const instance = yield* workflow.get("report-123");
995 * const events = yield* instance.subscribe({ filter: ["workflow_queued"] }).pipe(
996 * Stream.take(1),
997 * Stream.runCollect,
998 * );
999 * yield* instance.delete();
1000 * const result = yield* workflow.deleteBatch(["report-456", "report-789"]);
1001 * // result.deleted contains successful IDs; result.errors contains per-instance failures.
1002 * ```
1003 *
1004 * @resource
1005 * @product Workflows
1006 * @category Workers & Compute
1007 */
1008export const Workflow: WorkflowClass = taggedFunction(WorkflowScope, ((
1009 ...args:
1010 | []
1011 | [name: string, impl: Effect.Effect<WorkflowImpl<any, any, unknown>>]
1012 | [
1013 name: string,
1014 props: WorkflowProps,
1015 impl: Effect.Effect<WorkflowImpl<any, any, unknown>>,
1016 ]
1017 | [name: string, props?: WorkflowRefProps]
1018) => {
1019 if (args.length === 0) {
1020 return Workflow;
1021 }
1022 const [name, second, third] = args;
1023 const impl = Effect.isEffect(second)
1024 ? second
1025 : Effect.isEffect(third)
1026 ? third
1027 : undefined;
1028 if (impl === undefined) {
1029 // Props-only (async) reference form: returns a plain `WorkflowLike` that an
1030 // async Worker binds via `env`. `WorkerAsyncBindings` emits the `workflow`
1031 // binding and drives `putWorkflow` for locally-hosted workflows.
1032 const refProps = second as WorkflowRefProps | undefined;
1033 return {
1034 kind: TypeId,
1035 name,
1036 workflowName: refProps?.workflowName,
1037 className: refProps?.className ?? name,
1038 scriptName: refProps?.scriptName,
1039 limits: refProps?.limits,
1040 schedules: refProps?.schedules,
1041 } satisfies WorkflowLike;
1042 }
1043 const props = Effect.isEffect(second) ? undefined : (second as WorkflowProps);
1044 return effectClass(
1045 Effect.gen(function* () {
1046 const worker = yield* Worker;
1047
1048 const workflow = yield* WorkflowResource(name, {
1049 workflowName: props?.workflowName,
1050 className: name,
1051 scriptName:
1052 props?.workflowName === undefined ? worker.workerName : undefined,
1053 limits: props?.limits,
1054 schedules: props?.schedules,
1055 });
1056 // Keep named identities resolvable for the engine's cold-adoption probe.
1057 if (props?.workflowName !== undefined) {
1058 yield* workflow.bind`${worker}`({ scriptName: worker.workerName });
1059 }
1060
1061 // Add the workflow binding to the Worker metadata
1062 yield* worker.bind`${name}`({
1063 bindings: [
1064 {
1065 type: "workflow",
1066 name,
1067 workflowName: props?.workflowName ?? workflow.workflowName,
1068 className: name,
1069 },
1070 ],
1071 });
1072
1073 const services = yield* Effect.context<Effect.Services<typeof impl>>();
1074
1075 const binding = yield* Effect.all([
1076 WorkerEnvironment,
1077 ALCHEMY_PHASE,
1078 ]).pipe(
1079 Effect.flatMap(([env, phase]) => {
1080 if (env === undefined || phase === "plan") {
1081 return Effect.succeed(undefined as any);
1082 }
1083 const wf = env[name];
1084 if (!wf) {
1085 return Effect.die(new Error(`Workflow '${name}' not found in env`));
1086 }
1087 return Effect.succeed(wf);
1088 }),
1089 );
1090
1091 const self: WorkflowHandle<any, any> = {
1092 Type: TypeId,
1093 name,
1094 create: (options?: WorkflowInstanceCreateOptions<any>) =>
1095 Effect.tryPromise(() => binding.create(options)).pipe(
1096 Effect.map(wrapInstance),
1097 Effect.orDie,
1098 ),
1099 createBatch: (batch: WorkflowInstanceCreateOptions<any>[]) =>
1100 Effect.tryPromise(
1101 () => binding.createBatch(batch) as Promise<any[]>,
1102 ).pipe(
1103 Effect.map((instances: any[]) => instances.map(wrapInstance)),
1104 Effect.orDie,
1105 ),
1106 deleteBatch: (instanceIds) =>
1107 Effect.tryPromise(
1108 () =>
1109 binding.deleteBatch(
1110 instanceIds,
1111 ) as Promise<WorkflowBatchDeleteResult>,
1112 ).pipe(Effect.orDie),
1113 get: (instanceId: string) =>
1114 Effect.tryPromise(() => binding.get(instanceId)).pipe(
1115 Effect.map(wrapInstance),
1116 Effect.orDie,
1117 ),
1118 };
1119
1120 const fn = yield* impl.pipe(
1121 Effect.provideService(WorkflowScope, self as any),
1122 );
1123
1124 yield* worker.export(name, {
1125 kind: "workflow",
1126 make: (env: unknown) =>
1127 Effect.succeed(((input: unknown) =>
1128 fn(input).pipe(
1129 Effect.provideService(
1130 WorkerEnvironment,
1131 env as Record<string, any>,
1132 ),
1133 )) as WorkflowImpl<any, any, unknown>).pipe(
1134 Effect.provideContext(services),
1135 ),
1136 } satisfies WorkflowExport);
1137
1138 return self;
1139 }),
1140 );
1141}) as any);
1142
1143Workflow.ref = (id, options) => WorkflowResource.ref(id, options);
1144
1145// ---------------------------------------------------------------------------
1146// WorkflowResource -- manages the Cloudflare Workflows API lifecycle
1147// ---------------------------------------------------------------------------
1148
1149export interface WorkflowResourceProps {
1150 /**
1151 * Account-global Workflow name. If omitted, a deterministic name is derived
1152 * from `scriptName` and `className`.
1153 *
1154 * @internal
1155 */
1156 workflowName?: string;
1157 className: string;
1158 /** Hosting Worker script, supplied here or through a host binding. */
1159 scriptName?: string;
1160 limits?: WorkflowLimits;
1161 /**
1162 * Cron expressions that create a new Workflow instance on each match.
1163 * Pass an empty array to remove all schedules.
1164 */
1165 schedules?: string[];
1166}
1167
1168export interface WorkflowResourceAttrs {
1169 workflowId: string;
1170 workflowName: string;
1171 className: string;
1172 scriptName: string;
1173 accountId: string;
1174 /**
1175 * Cron expressions currently attached to this Workflow.
1176 */
1177 schedules: string[];
1178}
1179
1180const WorkflowResourceTypeId = "Cloudflare.Workflow";
1181
1182export interface WorkflowResource extends Resource<
1183 typeof WorkflowResourceTypeId,
1184 WorkflowResourceProps,
1185 WorkflowResourceAttrs,
1186 { scriptName: string }
1187> {}
1188
1189export const WorkflowResource = Resource<WorkflowResource>(
1190 WorkflowResourceTypeId,
1191);
1192
1193const getWorkflowOrUndefined = (accountId: string, workflowName: string) =>
1194 workflows
1195 .getWorkflow({ accountId, workflowName })
1196 .pipe(Effect.catchTag("WorkflowNotFound", () => Effect.succeed(undefined)));
1197
1198export const ProviderLive = () =>
1199 Provider.succeed(WorkflowResource, {
1200 // `workflowId` is stable across live updates: `putWorkflow` is a
1201 // PUT-as-upsert keyed by the account-global `workflowName`, so re-putting
1202 // the same name (className/scriptName changes) preserves the workflow's
1203 // id. The dev→deploy id change that used to force `workflowId` out of
1204 // this list is now an engine-orchestrated mode-switch REPLACEMENT (the
1205 // persisted `providerMode` stamp differs), so the live provider never
1206 // sees a `dev:` id.
1207 stables: ["workflowId", "accountId", "workflowName"],
1208 // Workflows are account-scoped. Enumerate every workflow in the account
1209 // via the paginated list API and hydrate each into the same Attributes
1210 // shape `reconcile` returns (id/name/className/scriptName are all on the
1211 // list item, so no per-item get is needed).
1212 list: () =>
1213 Effect.gen(function* () {
1214 const { accountId } = yield* yield* CloudflareEnvironment;
1215 return yield* workflows.listWorkflows.pages({ accountId }).pipe(
1216 Stream.runCollect,
1217 Effect.map((chunk) =>
1218 Array.from(chunk).flatMap((page) =>
1219 (page.result ?? []).map((wf) => ({
1220 workflowId: wf.id,
1221 workflowName: wf.name,
1222 // `className`/`scriptName` can be null/absent in the list
1223 // payload on some accounts — fall back so listing succeeds.
1224 className: wf.className ?? "",
1225 scriptName: wf.scriptName ?? "",
1226 accountId,
1227 schedules: fromObservedSchedules(wf.schedules),
1228 })),
1229 ),
1230 ),
1231 );
1232 }),
1233 diff: Effect.fn(function* ({
1234 id,
1235 olds,
1236 news,
1237 output,
1238 oldBindings,
1239 newBindings,
1240 }) {
1241 const { accountId } = yield* yield* CloudflareEnvironment;
1242 if (output?.accountId !== undefined && output.accountId !== accountId) {
1243 return { action: "replace" } as const;
1244 }
1245
1246 // The host script may be unresolved even when the physical name is known.
1247 const explicitName =
1248 "workflowName" in news && isResolved(news.workflowName)
1249 ? news.workflowName
1250 : undefined;
1251 const oldName =
1252 output?.workflowName ??
1253 olds.workflowName ??
1254 (olds.scriptName === undefined
1255 ? undefined
1256 : yield* generateWorkflowName(olds.scriptName, olds.className));
1257 // Omitting the name preserves the deployed identity, including legacy names.
1258 if (explicitName !== undefined && explicitName !== oldName) {
1259 const existing = yield* getWorkflowOrUndefined(accountId, explicitName);
1260 if (existing !== undefined) {
1261 return yield* new OwnedBySomeoneElse({
1262 message: `Cannot replace Workflow '${oldName}' with occupied name '${explicitName}'. Choose an unused name.`,
1263 resourceType: WorkflowResourceTypeId,
1264 logicalId: id,
1265 physicalName: explicitName,
1266 });
1267 }
1268 return { action: "replace" } as const;
1269 }
1270 if (
1271 !isResolved(newBindings) ||
1272 havePropsChanged(oldBindings, newBindings)
1273 ) {
1274 return { action: "update" } as const;
1275 }
1276 }),
1277 read: Effect.fn(function* ({ output, olds }) {
1278 const { accountId } = yield* yield* CloudflareEnvironment;
1279 const workflowName =
1280 output?.workflowName ??
1281 olds?.workflowName ??
1282 (olds?.scriptName === undefined
1283 ? undefined
1284 : yield* generateWorkflowName(olds.scriptName, olds.className));
1285 if (workflowName === undefined) return undefined;
1286
1287 const acct = output?.accountId ?? accountId;
1288 const workflow = yield* getWorkflowOrUndefined(acct, workflowName);
1289 if (workflow === undefined) return undefined;
1290 const attrs = {
1291 workflowId: workflow.id,
1292 workflowName: workflow.name,
1293 className: workflow.className,
1294 scriptName: workflow.scriptName,
1295 accountId: acct,
1296 schedules: fromObservedSchedules(workflow.schedules),
1297 };
1298 // Explicit names carry no ownership marker; cold reads require adoption.
1299 return output === undefined && olds?.workflowName !== undefined
1300 ? Unowned(attrs)
1301 : attrs;
1302 }),
1303 reconcile: Effect.fn(function* ({ id, news, output, bindings }) {
1304 const { accountId } = yield* yield* CloudflareEnvironment;
1305 const scriptName = yield* resolveWorkflowScriptName(news, bindings);
1306 const acct = output?.accountId ?? accountId;
1307 const workflowName =
1308 news.workflowName ??
1309 output?.workflowName ??
1310 (yield* generateWorkflowName(scriptName, news.className));
1311
1312 if (
1313 news.workflowName !== undefined &&
1314 output?.workflowName !== undefined &&
1315 news.workflowName !== output.workflowName
1316 ) {
1317 return yield* Effect.fail(
1318 new Error(
1319 `Workflow physical name changed from '${output.workflowName}' to '${news.workflowName}' during an in-place update; a replacement is required.`,
1320 ),
1321 );
1322 }
1323
1324 const existing = yield* getWorkflowOrUndefined(acct, workflowName);
1325 // Re-check occupied names at apply time; replacement is not adoption.
1326 if (
1327 news.workflowName !== undefined &&
1328 output === undefined &&
1329 existing !== undefined
1330 ) {
1331 return yield* new OwnedBySomeoneElse({
1332 message:
1333 `Cannot create Workflow '${workflowName}': an existing ` +
1334 "Workflow has that account-global name. Choose an unused name, " +
1335 "or adopt it into a resource with no prior state by re-planning " +
1336 "with --adopt.",
1337 resourceType: WorkflowResourceTypeId,
1338 logicalId: id,
1339 physicalName: workflowName,
1340 });
1341 }
1342 // PUT clears omitted schedules; preserve observed state unless explicitly set.
1343 const schedules =
1344 news.schedules ?? fromObservedSchedules(existing?.schedules);
1345
1346 yield* Effect.logInfo(`Cloudflare Workflow reconcile: ${workflowName}`);
1347 // Cloudflare's `putWorkflow` is a true PUT-as-upsert: identical
1348 // payloads converge to the same state and a missing workflow is
1349 // created on the spot.
1350 const result = yield* workflows.putWorkflow({
1351 accountId: acct,
1352 workflowName,
1353 className: news.className,
1354 scriptName,
1355 limits: news.limits,
1356 schedules: toPutSchedules(schedules),
1357 });
1358 return {
1359 workflowId: result.id,
1360 workflowName: result.name,
1361 className: result.className,
1362 scriptName: result.scriptName,
1363 accountId: acct,
1364 schedules,
1365 };
1366 }),
1367 delete: Effect.fn(function* ({ output }) {
1368 yield* Effect.logInfo(
1369 `Cloudflare Workflow delete: ${output.workflowName}`,
1370 );
1371 yield* workflows
1372 .deleteWorkflow({
1373 accountId: output.accountId,
1374 workflowName: output.workflowName,
1375 })
1376 .pipe(Effect.catchTag("WorkflowNotFound", () => Effect.void));
1377 }),
1378 });
1379
1380/**
1381 * Local (dev) provider — the workflow is purely virtual: a `dev:` id keyed
1382 * into the local workerd workflow engine. The host worker's `workflow`
1383 * binding is lowered onto the local runtime's Workflow Engine DO by
1384 * `LocalWorkerProvider` (`Workflows.local(...)`), so no runtime layer is
1385 * needed here; instance state persists under the worker's local storage.
1386 */
1387export const ProviderLocal = () =>
1388 Provider.succeed(WorkflowResource, {
1389 stables: ["accountId"],
1390 diff: Effect.fn(function* ({ news, output, oldBindings, newBindings }) {
1391 const accountId = yield* localAccountId;
1392 if (!output?.workflowId) return { action: "update" } as const;
1393 if (output.accountId !== accountId) {
1394 return { action: "replace" } as const;
1395 }
1396 // The host script may be unresolved even when the physical name is known.
1397 const explicitName =
1398 "workflowName" in news && isResolved(news.workflowName)
1399 ? news.workflowName
1400 : undefined;
1401 if (explicitName !== undefined && explicitName !== output.workflowName) {
1402 return { action: "replace" } as const;
1403 }
1404 if (
1405 !isResolved(newBindings) ||
1406 havePropsChanged(oldBindings, newBindings)
1407 ) {
1408 return { action: "update" } as const;
1409 }
1410 // Fall through to the engine's default prop diff (className /
1411 // scriptName changes update in place).
1412 }),
1413 read: Effect.fn(function* ({ output }) {
1414 // Purely virtual — the persisted state row is the source of truth.
1415 return output ?? undefined;
1416 }),
1417 reconcile: Effect.fn(function* ({ news, output, bindings }) {
1418 const accountId = yield* localAccountId;
1419 const scriptName = yield* resolveWorkflowScriptName(news, bindings);
1420 if (
1421 news.workflowName !== undefined &&
1422 output?.workflowName !== undefined &&
1423 news.workflowName !== output.workflowName
1424 ) {
1425 return yield* Effect.fail(
1426 new Error(
1427 `Workflow physical name changed from '${output.workflowName}' to '${news.workflowName}' during an in-place update; a replacement is required.`,
1428 ),
1429 );
1430 }
1431 return {
1432 workflowId: output?.workflowId ?? generateLocalId(),
1433 workflowName:
1434 news.workflowName ??
1435 output?.workflowName ??
1436 (yield* generateWorkflowName(scriptName, news.className)),
1437 className: news.className,
1438 scriptName,
1439 accountId: output?.accountId ?? accountId,
1440 schedules: news.schedules ?? output?.schedules ?? [],
1441 };
1442 }),
1443 delete: Effect.fn(function* () {
1444 // The simulator's on-disk instance state lives under the local worker's
1445 // storage; dropping the state row is enough — orphaned data is
1446 // reclaimed with `.alchemy`.
1447 }),
1448 });
1449
1450export const WorkflowProvider = () =>
1451 ProviderLayer.dual(WorkflowResource, {
1452 local: () => ProviderLocal(),
1453 live: () => ProviderLive(),
1454 });
1455
1456// ---------------------------------------------------------------------------
1457// Helpers
1458// ---------------------------------------------------------------------------
1459
1460const resolveWorkflowScriptName = (
1461 props: WorkflowResourceProps,
1462 bindings: ReadonlyArray<{ data: { scriptName: string } }>,
1463) => {
1464 const scripts = [
1465 ...new Set(
1466 [
1467 props.scriptName,
1468 ...bindings.map((binding) => binding.data.scriptName),
1469 ].filter((name) => name !== undefined),
1470 ),
1471 ];
1472 return scripts.length === 1
1473 ? Effect.succeed(scripts[0]!)
1474 : Effect.fail(
1475 new Error("Workflow requires exactly one hosting Worker script"),
1476 );
1477};
1478
1479const toPutSchedules = (
1480 schedules: string[],
1481): workflows.UpdateRequestSchedulesList => schedules.map((cron) => ({ cron }));
1482
1483const fromObservedSchedules = (
1484 schedules?: ReadonlyArray<{ cron: string }> | null,
1485): string[] => (schedules ?? []).map((s) => s.cron);
1486
1487const wrapInstance = <Result>(raw: any): WorkflowInstance<Result> => ({
1488 id: raw.id,
1489 status: () =>
1490 Effect.tryPromise(() => raw.status()).pipe(
1491 Effect.map((s: any) => ({
1492 status: s.status as string,
1493 output: s.output as Result,
1494 error: s.error,
1495 rollback: s.rollback,
1496 })),
1497 Effect.orDie,
1498 ),
1499 pause: () => Effect.tryPromise(() => raw.pause()).pipe(Effect.orDie),
1500 resume: () => Effect.tryPromise(() => raw.resume()).pipe(Effect.orDie),
1501 restart: (options?: WorkflowInstanceRestartOptions) =>
1502 Effect.tryPromise(() => raw.restart(options)).pipe(Effect.orDie),
1503 terminate: () => Effect.tryPromise(() => raw.terminate()).pipe(Effect.orDie),
1504 delete: () => Effect.tryPromise(() => raw.delete()).pipe(Effect.orDie),
1505 subscribe: (options) =>
1506 Stream.unwrap(
1507 Effect.acquireRelease(
1508 Effect.tryPromise(
1509 () =>
1510 raw.subscribe(options) as Promise<
1511 runtime.WorkflowInstanceSubscription & Disposable
1512 >,
1513 ).pipe(Effect.orDie),
1514 (subscription) => Effect.sync(() => subscription[Symbol.dispose]()),
1515 ).pipe(
1516 Effect.map((subscription) =>
1517 Stream.fromAsyncIterable(
1518 {
1519 [Symbol.asyncIterator]: () => ({
1520 next: () => subscription.next(),
1521 }),
1522 },
1523 (error) => error,
1524 ).pipe(Stream.orDie),
1525 ),
1526 ),
1527 ),
1528 sendEvent: <Event = unknown>(event: WorkflowInstanceEvent<Event>) =>
1529 Effect.tryPromise(() => raw.sendEvent(event)).pipe(Effect.orDie),
1530});